You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by vi...@apache.org on 2013/11/04 18:09:02 UTC
[2/3] git commit: ACCUMULO-1679 - removing TFile and Chunk
ACCUMULO-1679 - removing TFile and Chunk
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/ef6ba24c
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/ef6ba24c
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/ef6ba24c
Branch: refs/heads/master
Commit: ef6ba24c71e3b15fa4944b3682a1c95d1eb6ba1f
Parents: 61d00d4
Author: John Vines <jv...@gmail.com>
Authored: Tue Oct 29 17:10:47 2013 -0400
Committer: John Vines <jv...@gmail.com>
Committed: Mon Nov 4 12:00:07 2013 -0500
----------------------------------------------------------------------
.../accumulo/core/file/rfile/bcfile/BCFile.java | 469 ++--
.../accumulo/core/file/rfile/bcfile/Chunk.java | 418 ----
.../core/file/rfile/bcfile/Compression.java | 17 +-
.../accumulo/core/file/rfile/bcfile/TFile.java | 2030 ------------------
.../core/file/rfile/bcfile/TFileDumper.java | 258 ---
5 files changed, 247 insertions(+), 2945 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ef6ba24c/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/BCFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/BCFile.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/BCFile.java
index 2d9f6af..5b416cc 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/BCFile.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/BCFile.java
@@ -64,14 +64,25 @@ public final class BCFile {
static final Version API_VERSION = new Version((short) 2, (short) 0);
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, 256 * 1024);
+ }
+
/**
* Prevent the instantiation of BCFile objects.
*/
private BCFile() {
// nothing
}
-
+
/**
* BCFile writer, the entry point for creating a new BCFile.
*/
@@ -93,7 +104,6 @@ public final class BCFile {
// reusable buffers.
private BytesWritable fsOutputBuffer;
-
/**
* Call-back interface to register a block after a block is closed.
*/
@@ -110,7 +120,7 @@ public final class BCFile {
*/
public void register(long raw, long offsetStart, long offsetEnd);
}
-
+
/**
* Intermediate class that maintain the state of a Writable Compression Block.
*/
@@ -123,39 +133,42 @@ public final class BCFile {
private final long posStart;
private final SimpleBufferedOutputStream fsBufferedOutput;
private OutputStream out;
-
+
/**
* @param compressionAlgo
* The compression algorithm to be used to for compression.
- * @param cryptoModule the module to use to obtain cryptographic streams
- * @param cryptoParams TODO
+ * @param cryptoModule
+ * the module to use to obtain cryptographic streams
+ * @param cryptoParams
* @throws IOException
*/
- public WBlockState(Algorithm compressionAlgo, FSDataOutputStream fsOut, BytesWritable fsOutputBuffer, Configuration conf, CryptoModule cryptoModule, CryptoModuleParameters cryptoParams) throws IOException {
+ public WBlockState(Algorithm compressionAlgo, FSDataOutputStream fsOut, BytesWritable fsOutputBuffer, Configuration conf, CryptoModule cryptoModule,
+ CryptoModuleParameters cryptoParams) throws IOException {
this.compressAlgo = compressionAlgo;
this.fsOut = fsOut;
this.posStart = fsOut.getPos();
-
- fsOutputBuffer.setCapacity(TFile.getFSOutputBufferSize(conf));
+
+ fsOutputBuffer.setCapacity(getFSOutputBufferSize(conf));
this.fsBufferedOutput = new SimpleBufferedOutputStream(this.fsOut, fsOutputBuffer.getBytes());
-
- // *This* is very important. Without this, when the crypto stream is closed (in order to flush its last bytes),
- // the underlying RFile stream will *also* be closed, and that's undesirable as the cipher stream is closed for
+
+ // *This* is very important. Without this, when the crypto stream is closed (in order to flush its last bytes),
+ // the underlying RFile stream will *also* be closed, and that's undesirable as the cipher stream is closed for
// every block written.
cryptoParams.setCloseUnderylingStreamAfterCryptoStreamClose(false);
-
- // *This* is also very important. We don't want the underlying stream messed with.
+
+ // *This* is also very important. We don't want the underlying stream messed with.
cryptoParams.setRecordParametersToStream(false);
-
+
// It is also important to make sure we get a new initialization vector on every call in here,
// so set any existing one to null, in case we're reusing a parameters object for its RNG or other bits
cryptoParams.setInitializationVector(null);
-
+
// Initialize the cipher including generating a new IV
cryptoParams = cryptoModule.initializeCipher(cryptoParams);
-
- // Write the init vector in plain text, uncompressed, to the output stream. Due to the way the streams work out, there's no good way to write this compressed, but it's pretty small.
+
+ // Write the init vector in plain text, uncompressed, to the output stream. Due to the way the streams work out, there's no good way to write this
+ // compressed, but it's pretty small.
DataOutputStream tempDataOutputStream = new DataOutputStream(fsBufferedOutput);
// Init vector might be null if the underlying cipher does not require one (NullCipher being a good example)
@@ -165,20 +178,19 @@ public final class BCFile {
} else {
// Do nothing
}
-
- // Initialize the cipher stream and get the IV
+
+ // Initialize the cipher stream and get the IV
cryptoParams.setPlaintextOutputStream(tempDataOutputStream);
cryptoParams = cryptoModule.getEncryptingOutputStream(cryptoParams);
-
+
if (cryptoParams.getEncryptedOutputStream() == tempDataOutputStream) {
this.cipherOut = fsBufferedOutput;
- } else {
+ } else {
this.cipherOut = cryptoParams.getEncryptedOutputStream();
}
-
-
+
this.compressor = compressAlgo.getCompressor();
-
+
try {
this.out = compressionAlgo.createCompressionStream(cipherOut, compressor, 0);
} catch (IOException e) {
@@ -186,7 +198,7 @@ public final class BCFile {
throw e;
}
}
-
+
/**
* Get the output stream for BlockAppender's consumption.
*
@@ -195,7 +207,7 @@ public final class BCFile {
OutputStream getOutputStream() {
return out;
}
-
+
/**
* Get the current position in file.
*
@@ -205,11 +217,11 @@ public final class BCFile {
long getCurrentPos() throws IOException {
return fsOut.getPos() + fsBufferedOutput.size();
}
-
+
long getStartPos() {
return posStart;
}
-
+
/**
* Current size of compressed data.
*
@@ -219,7 +231,7 @@ public final class BCFile {
long ret = getCurrentPos() - posStart;
return ret;
}
-
+
/**
* Finishing up the current block.
*/
@@ -227,18 +239,18 @@ public final class BCFile {
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
+ // 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 {
@@ -247,7 +259,7 @@ public final class BCFile {
}
}
}
-
+
/**
* Access point to stuff data into a block.
*
@@ -256,7 +268,7 @@ public final class BCFile {
private final BlockRegister blockRegister;
private final WBlockState wBlkState;
private boolean closed = false;
-
+
/**
* Constructor
*
@@ -270,7 +282,7 @@ public final class BCFile {
this.blockRegister = register;
this.wBlkState = wbs;
}
-
+
/**
* Get the raw size of the block.
*
@@ -283,7 +295,7 @@ public final class BCFile {
*/
return size() & 0x00000000ffffffffL;
}
-
+
/**
* Get the compressed size of the block in progress.
*
@@ -294,17 +306,17 @@ public final class BCFile {
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.
*/
@@ -322,9 +334,9 @@ public final class BCFile {
closed = true;
blkInProgress = false;
}
- }
+ }
}
-
+
/**
* Constructor
*
@@ -339,16 +351,16 @@ public final class BCFile {
if (fout.getPos() != 0) {
throw new IOException("Output file not at zero offset.");
}
-
+
this.out = fout;
this.conf = conf;
dataIndex = new DataIndex(compressionName, trackDataBlocks);
metaIndex = new MetaIndex();
fsOutputBuffer = new BytesWritable();
Magic.write(fout);
-
+
// Set up crypto-related detail, including secret key generation and encryption
-
+
@SuppressWarnings("deprecation")
AccumuloConfiguration accumuloConfiguration = AccumuloConfiguration.getSiteConfiguration();
this.cryptoConf = accumuloConfiguration.getAllPropertiesWithPrefix(Property.CRYPTO_PREFIX);
@@ -362,15 +374,14 @@ public final class BCFile {
this.cryptoParams = new BCFileCryptoModuleParameters();
CryptoModuleFactory.fillParamsObjectFromStringMap(cryptoParams, cryptoConf);
this.cryptoParams = (BCFileCryptoModuleParameters) cryptoModule.generateNewRandomSessionKey(cryptoParams);
-
- this.secretKeyEncryptionStrategy = CryptoModuleFactory.getSecretKeyEncryptionStrategy(accumuloConfiguration);
+
+ this.secretKeyEncryptionStrategy = CryptoModuleFactory.getSecretKeyEncryptionStrategy(accumuloConfiguration);
this.cryptoParams = (BCFileCryptoModuleParameters) secretKeyEncryptionStrategy.encryptSecretKey(cryptoParams);
-
-
+
// secretKeyEncryptionStrategy.encryptSecretKey(cryptoParameters);
-
+
}
-
+
/**
* Close the BCFile Writer. Attempting to use the Writer after calling <code>close</code> is not allowed and may lead to undetermined results.
*/
@@ -378,13 +389,13 @@ public final class BCFile {
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 {
@@ -392,18 +403,17 @@ public final class BCFile {
} finally {
appender.close();
}
-
+
long offsetIndexMeta = out.getPos();
metaIndex.write(out);
- if (cryptoParams.getAlgorithmName() == null ||
- cryptoParams.getAlgorithmName().equals(Property.CRYPTO_CIPHER_SUITE.getDefaultValue())) {
+ if (cryptoParams.getAlgorithmName() == null || cryptoParams.getAlgorithmName().equals(Property.CRYPTO_CIPHER_SUITE.getDefaultValue())) {
out.writeLong(offsetIndexMeta);
API_VERSION_1.write(out);
} else {
long offsetCryptoParameters = out.getPos();
cryptoParams.write(out);
-
+
// Meta Index, crypto params offsets and the trailing section are written out directly.
out.writeLong(offsetIndexMeta);
out.writeLong(offsetCryptoParameters);
@@ -417,20 +427,20 @@ public final class BCFile {
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, cryptoModule, cryptoParams);
BlockAppender ba = new BlockAppender(mbr, wbs);
@@ -438,7 +448,7 @@ public final class BCFile {
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.
@@ -455,7 +465,7 @@ public final class BCFile {
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
@@ -471,7 +481,7 @@ public final class BCFile {
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.
@@ -483,36 +493,36 @@ public final class BCFile {
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, cryptoModule, cryptoParams);
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;
}
-
+
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.
*
@@ -521,13 +531,13 @@ public final class BCFile {
DataBlockRegister() {
// do nothing
}
-
+
public void register(long raw, long begin, long end) {
dataIndex.addBlockRegion(new BlockRegion(begin, end - begin, raw));
}
}
}
-
+
private static class BCFileCryptoModuleParameters extends CryptoModuleParameters {
public void write(DataOutput out) throws IOException {
@@ -537,43 +547,40 @@ public final class BCFile {
out.writeUTF(key);
out.writeUTF(getAllOptions().get(key));
}
-
+
// Write the opaque ID
out.writeUTF(getOpaqueKeyEncryptionKeyID());
-
+
// Write the encrypted secret key
out.writeInt(getEncryptedKey().length);
out.write(getEncryptedKey());
-
+
}
-
+
public void read(DataInput in) throws IOException {
-
+
Map<String,String> optionsFromFile = new HashMap<String,String>();
-
-
-
+
int numContextEntries = in.readInt();
for (int i = 0; i < numContextEntries; i++) {
optionsFromFile.put(in.readUTF(), in.readUTF());
}
-
+
CryptoModuleFactory.fillParamsObjectFromStringMap(this, optionsFromFile);
-
+
// Read opaque key encryption ID
setOpaqueKeyEncryptionKeyID(in.readUTF());
-
+
// Read encrypted secret key
int encryptedSecretKeyLength = in.readInt();
byte[] encryptedSecretKey = new byte[encryptedSecretKeyLength];
in.readFully(encryptedSecretKey);
setEncryptedKey(encryptedSecretKey);
-
+
}
-
}
-
+
/**
* BCFile Reader, interface to read the file's data and meta blocks.
*/
@@ -589,7 +596,7 @@ public final class BCFile {
private BCFileCryptoModuleParameters cryptoParams;
private CryptoModule cryptoModule;
private SecretKeyEncryptionStrategy secretKeyEncryptionStrategy;
-
+
/**
* Intermediate class that maintain the state of a Readable Compression Block.
*/
@@ -598,45 +605,45 @@ public final class BCFile {
private Decompressor decompressor;
private final BlockRegion region;
private final InputStream in;
-
- public RBlockState(Algorithm compressionAlgo, FSDataInputStream fsin, BlockRegion region, Configuration conf, CryptoModule cryptoModule, Version bcFileVersion, CryptoModuleParameters cryptoParams) throws IOException {
+
+ public RBlockState(Algorithm compressionAlgo, FSDataInputStream fsin, BlockRegion region, Configuration conf, CryptoModule cryptoModule,
+ Version bcFileVersion, CryptoModuleParameters cryptoParams) throws IOException {
this.compressAlgo = compressionAlgo;
this.region = region;
this.decompressor = compressionAlgo.getDecompressor();
-
- BoundedRangeFileInputStream boundedRangeFileInputStream = new BoundedRangeFileInputStream(fsin, this.region.getOffset(), this.region.getCompressedSize());
+
+ BoundedRangeFileInputStream boundedRangeFileInputStream = new BoundedRangeFileInputStream(fsin, this.region.getOffset(),
+ this.region.getCompressedSize());
InputStream inputStreamToBeCompressed = boundedRangeFileInputStream;
-
+
if (cryptoParams != null && cryptoModule != null) {
DataInputStream tempDataInputStream = new DataInputStream(boundedRangeFileInputStream);
// Read the init vector from the front of the stream before initializing the cipher stream
-
+
int ivLength = tempDataInputStream.readInt();
byte[] initVector = new byte[ivLength];
tempDataInputStream.readFully(initVector);
-
+
cryptoParams.setInitializationVector(initVector);
cryptoParams.setEncryptedInputStream(boundedRangeFileInputStream);
-
-
+
// These two flags mirror those in WBlockState, and are very necessary to set in order that the underlying stream be written and handled
// correctly.
cryptoParams.setCloseUnderylingStreamAfterCryptoStreamClose(false);
cryptoParams.setRecordParametersToStream(false);
-
-
+
cryptoParams = cryptoModule.getDecryptingInputStream(cryptoParams);
inputStreamToBeCompressed = cryptoParams.getPlaintextInputStream();
}
-
+
try {
- this.in = compressAlgo.createDecompressionStream(inputStreamToBeCompressed, decompressor, TFile.getFSInputBufferSize(conf));
+ this.in = compressAlgo.createDecompressionStream(inputStreamToBeCompressed, decompressor, getFSInputBufferSize(conf));
} catch (IOException e) {
compressAlgo.returnDecompressor(decompressor);
throw e;
}
}
-
+
/**
* Get the output stream for BlockAppender's consumption.
*
@@ -645,15 +652,15 @@ public final class BCFile {
public InputStream getInputStream() {
return in;
}
-
+
public String getCompressionName() {
return compressAlgo.getName();
}
-
+
public BlockRegion getBlockRegion() {
return region;
}
-
+
public void finish() throws IOException {
try {
in.close();
@@ -663,19 +670,19 @@ public final class BCFile {
}
}
}
-
+
/**
* 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.
*/
@@ -692,7 +699,7 @@ public final class BCFile {
closed = true;
}
}
-
+
/**
* Get the name of the compression algorithm used to compress the block.
*
@@ -701,7 +708,7 @@ public final class BCFile {
public String getCompressionName() {
return rBlkState.getCompressionName();
}
-
+
/**
* Get the uncompressed size of the block.
*
@@ -710,7 +717,7 @@ public final class BCFile {
public long getRawSize() {
return rBlkState.getBlockRegion().getRawSize();
}
-
+
/**
* Get the compressed size of the block.
*
@@ -719,7 +726,7 @@ public final class BCFile {
public long getCompressedSize() {
return rBlkState.getBlockRegion().getCompressedSize();
}
-
+
/**
* Get the starting position of the block in the file.
*
@@ -729,7 +736,7 @@ public final class BCFile {
return rBlkState.getBlockRegion().getOffset();
}
}
-
+
/**
* Constructor
*
@@ -740,11 +747,10 @@ public final class BCFile {
* @throws IOException
*/
public Reader(FSDataInputStream fin, long fileLength, Configuration conf) throws IOException {
-
+
this.in = fin;
this.conf = conf;
-
-
+
// Move the cursor to grab the version and the magic first
fin.seek(fileLength - Magic.size() - Version.size());
version = new Version(fin);
@@ -754,46 +760,45 @@ public final class BCFile {
if (!version.compatibleWith(BCFile.API_VERSION) && !version.equals(BCFile.API_VERSION_1)) {
throw new RuntimeException("Incompatible BCFile fileBCFileVersion.");
}
-
+
// Read the right number offsets based on version
long offsetIndexMeta = 0;
long offsetCryptoParameters = 0;
-
+
if (version.equals(API_VERSION_1)) {
- fin.seek(fileLength - Magic.size() - Version.size() - ( Long.SIZE / Byte.SIZE ) );
+ fin.seek(fileLength - Magic.size() - Version.size() - (Long.SIZE / Byte.SIZE));
offsetIndexMeta = fin.readLong();
-
+
} else {
- fin.seek(fileLength - Magic.size() - Version.size() - ( 2 * ( Long.SIZE / Byte.SIZE ) ));
+ fin.seek(fileLength - Magic.size() - Version.size() - (2 * (Long.SIZE / Byte.SIZE)));
offsetIndexMeta = fin.readLong();
offsetCryptoParameters = fin.readLong();
}
-
-
+
// read meta index
fin.seek(offsetIndexMeta);
metaIndex = new MetaIndex(fin);
// If they exist, read the crypto parameters
if (!version.equals(BCFile.API_VERSION_1)) {
-
+
@SuppressWarnings("deprecation")
AccumuloConfiguration accumuloConfiguration = AccumuloConfiguration.getSiteConfiguration();
-
+
// read crypto parameters
fin.seek(offsetCryptoParameters);
cryptoParams = new BCFileCryptoModuleParameters();
cryptoParams.read(fin);
-
+
this.cryptoModule = CryptoModuleFactory.getCryptoModule(cryptoParams.getAllOptions().get(Property.CRYPTO_MODULE_CLASS.getKey()));
-
- // TODO: Do I need this? Hmmm, maybe I do.
+
+ // TODO: Do I need this? Hmmm, maybe I do.
if (accumuloConfiguration.getBoolean(Property.CRYPTO_OVERRIDE_KEY_STRATEGY_WITH_CONFIGURED_STRATEGY)) {
Map<String,String> cryptoConfFromAccumuloConf = accumuloConfiguration.getAllPropertiesWithPrefix(Property.CRYPTO_PREFIX);
Map<String,String> instanceConf = accumuloConfiguration.getAllPropertiesWithPrefix(Property.INSTANCE_PREFIX);
-
+
cryptoConfFromAccumuloConf.putAll(instanceConf);
-
+
for (String name : cryptoParams.getAllOptions().keySet()) {
if (!name.equals(Property.CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS.getKey())) {
cryptoConfFromAccumuloConf.put(name, cryptoParams.getAllOptions().get(name));
@@ -801,21 +806,20 @@ public final class BCFile {
cryptoParams.setKeyEncryptionStrategyClass(cryptoConfFromAccumuloConf.get(Property.CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS.getKey()));
}
}
-
+
cryptoParams.setAllOptions(cryptoConfFromAccumuloConf);
}
-
+
this.secretKeyEncryptionStrategy = CryptoModuleFactory.getSecretKeyEncryptionStrategy(cryptoParams.getKeyEncryptionStrategyClass());
-
+
// This call should put the decrypted session key within the cryptoParameters object
cryptoParams = (BCFileCryptoModuleParameters) secretKeyEncryptionStrategy.decryptSecretKey(cryptoParams);
-
-
- //secretKeyEncryptionStrategy.decryptSecretKey(cryptoParameters);
+
+ // secretKeyEncryptionStrategy.decryptSecretKey(cryptoParameters);
} else {
LOG.trace("Found a version 1 file to read.");
}
-
+
// read data:BCFile.index, the data block index
BlockReader blockR = getMetaBlock(DataIndex.BLOCK_NAME);
try {
@@ -824,15 +828,15 @@ public final class BCFile {
blockR.close();
}
}
-
+
public Reader(CachableBlockFile.Reader cache, FSDataInputStream fin, long fileLength, Configuration conf) throws IOException {
this.in = fin;
this.conf = conf;
-
+
BlockRead cachedMetaIndex = cache.getCachedMetaBlock(META_NAME);
BlockRead cachedDataIndex = cache.getCachedMetaBlock(DataIndex.BLOCK_NAME);
BlockRead cachedCryptoParams = cache.getCachedMetaBlock(CRYPTO_BLOCK_NAME);
-
+
if (cachedMetaIndex == null || cachedDataIndex == null || cachedCryptoParams == null) {
// move the cursor to the beginning of the tail, containing: offset to the
// meta block index, version and magic
@@ -845,44 +849,42 @@ public final class BCFile {
if (!version.compatibleWith(BCFile.API_VERSION) && !version.equals(BCFile.API_VERSION_1)) {
throw new RuntimeException("Incompatible BCFile fileBCFileVersion.");
}
-
+
// Read the right number offsets based on version
long offsetIndexMeta = 0;
long offsetCryptoParameters = 0;
-
+
if (version.equals(API_VERSION_1)) {
- fin.seek(fileLength - Magic.size() - Version.size() - ( Long.SIZE / Byte.SIZE ) );
+ fin.seek(fileLength - Magic.size() - Version.size() - (Long.SIZE / Byte.SIZE));
offsetIndexMeta = fin.readLong();
-
+
} else {
- fin.seek(fileLength - Magic.size() - Version.size() - ( 2 * ( Long.SIZE / Byte.SIZE ) ));
+ fin.seek(fileLength - Magic.size() - Version.size() - (2 * (Long.SIZE / Byte.SIZE)));
offsetIndexMeta = fin.readLong();
offsetCryptoParameters = fin.readLong();
}
-
+
// read meta index
fin.seek(offsetIndexMeta);
metaIndex = new MetaIndex(fin);
-
+
// If they exist, read the crypto parameters
if (!version.equals(BCFile.API_VERSION_1) && cachedCryptoParams == null) {
-
+
@SuppressWarnings("deprecation")
AccumuloConfiguration accumuloConfiguration = AccumuloConfiguration.getSiteConfiguration();
-
// read crypto parameters
fin.seek(offsetCryptoParameters);
cryptoParams = new BCFileCryptoModuleParameters();
cryptoParams.read(fin);
-
-
+
if (accumuloConfiguration.getBoolean(Property.CRYPTO_OVERRIDE_KEY_STRATEGY_WITH_CONFIGURED_STRATEGY)) {
Map<String,String> cryptoConfFromAccumuloConf = accumuloConfiguration.getAllPropertiesWithPrefix(Property.CRYPTO_PREFIX);
Map<String,String> instanceConf = accumuloConfiguration.getAllPropertiesWithPrefix(Property.INSTANCE_PREFIX);
-
+
cryptoConfFromAccumuloConf.putAll(instanceConf);
-
+
for (String name : cryptoParams.getAllOptions().keySet()) {
if (!name.equals(Property.CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS.getKey())) {
cryptoConfFromAccumuloConf.put(name, cryptoParams.getAllOptions().get(name));
@@ -890,39 +892,38 @@ public final class BCFile {
cryptoParams.setKeyEncryptionStrategyClass(cryptoConfFromAccumuloConf.get(Property.CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS.getKey()));
}
}
-
+
cryptoParams.setAllOptions(cryptoConfFromAccumuloConf);
}
-
-
+
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
cryptoParams.write(dos);
dos.close();
cache.cacheMetaBlock(CRYPTO_BLOCK_NAME, baos.toByteArray());
-
+
this.cryptoModule = CryptoModuleFactory.getCryptoModule(cryptoParams.getAllOptions().get(Property.CRYPTO_MODULE_CLASS.getKey()));
this.secretKeyEncryptionStrategy = CryptoModuleFactory.getSecretKeyEncryptionStrategy(cryptoParams.getKeyEncryptionStrategyClass());
-
+
// This call should put the decrypted session key within the cryptoParameters object
// secretKeyEncryptionStrategy.decryptSecretKey(cryptoParameters);
-
+
cryptoParams = (BCFileCryptoModuleParameters) secretKeyEncryptionStrategy.decryptSecretKey(cryptoParams);
-
+
} else if (cachedCryptoParams != null) {
cryptoParams = new BCFileCryptoModuleParameters();
cryptoParams.read(cachedCryptoParams);
-
+
this.cryptoModule = CryptoModuleFactory.getCryptoModule(cryptoParams.getAllOptions().get(Property.CRYPTO_MODULE_CLASS.getKey()));
this.secretKeyEncryptionStrategy = CryptoModuleFactory.getSecretKeyEncryptionStrategy(cryptoParams.getKeyEncryptionStrategyClass());
-
+
// This call should put the decrypted session key within the cryptoParameters object
// secretKeyEncryptionStrategy.decryptSecretKey(cryptoParameters);
-
+
cryptoParams = (BCFileCryptoModuleParameters) secretKeyEncryptionStrategy.decryptSecretKey(cryptoParams);
-
+
}
-
+
if (cachedMetaIndex == null) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
@@ -930,14 +931,13 @@ public final class BCFile {
dos.close();
cache.cacheMetaBlock(META_NAME, baos.toByteArray());
}
-
+
// read data:BCFile.index, the data block index
if (cachedDataIndex == null) {
BlockReader blockR = getMetaBlock(DataIndex.BLOCK_NAME);
cachedDataIndex = cache.cacheMetaBlock(DataIndex.BLOCK_NAME, blockR);
}
-
-
+
try {
dataIndex = new DataIndex(cachedDataIndex);
} catch (IOException e) {
@@ -945,27 +945,26 @@ public final class BCFile {
throw e;
}
cachedDataIndex.close();
-
+
} else {
// We have cached versions of the metaIndex, dataIndex and cryptoParams objects.
// Use them to fill out this reader's members.
version = null;
-
+
metaIndex = new MetaIndex(cachedMetaIndex);
dataIndex = new DataIndex(cachedDataIndex);
cryptoParams = new BCFileCryptoModuleParameters();
cryptoParams.read(cachedCryptoParams);
-
+
this.cryptoModule = CryptoModuleFactory.getCryptoModule(cryptoParams.getAllOptions().get(Property.CRYPTO_MODULE_CLASS.getKey()));
this.secretKeyEncryptionStrategy = CryptoModuleFactory.getSecretKeyEncryptionStrategy(cryptoParams.getKeyEncryptionStrategyClass());
-
+
// This call should put the decrypted session key within the cryptoParameters object
cryptoParams = (BCFileCryptoModuleParameters) secretKeyEncryptionStrategy.decryptSecretKey(cryptoParams);
-
}
}
-
+
/**
* Get the name of the default compression algorithm.
*
@@ -974,7 +973,7 @@ public final class BCFile {
public String getDefaultCompressionName() {
return dataIndex.getDefaultCompressionAlgorithm().getName();
}
-
+
/**
* Get version of BCFile file being read.
*
@@ -983,7 +982,7 @@ public final class BCFile {
public Version getBCFileVersion() {
return version;
}
-
+
/**
* Get version of BCFile API.
*
@@ -992,14 +991,14 @@ public final class BCFile {
public Version getAPIVersion() {
return API_VERSION;
}
-
+
/**
* Finishing reading the BCFile. Release all resources.
*/
public void close() {
// nothing to be done now
}
-
+
/**
* Get the number of data blocks.
*
@@ -1008,7 +1007,7 @@ public final class BCFile {
public int getBlockCount() {
return dataIndex.getBlockRegionList().size();
}
-
+
/**
* Stream access to a Meta Block.
*
@@ -1024,11 +1023,11 @@ public final class BCFile {
if (imeBCIndex == null) {
throw new MetaBlockDoesNotExist("name=" + name);
}
-
+
BlockRegion region = imeBCIndex.getRegion();
return createReader(imeBCIndex.getCompressionAlgorithm(), region);
}
-
+
/**
* Stream access to a Data Block.
*
@@ -1041,21 +1040,21 @@ public final class BCFile {
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);
}
-
+
private BlockReader createReader(Algorithm compressAlgo, BlockRegion region) throws IOException {
RBlockState rbs = new RBlockState(compressAlgo, in, region, conf, cryptoModule, version, cryptoParams);
return new BlockReader(rbs);
}
-
+
/**
* Find the smallest Block index whose starting offset is greater than or equal to the specified offset.
*
@@ -1066,55 +1065,55 @@ public final class BCFile {
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.
*/
@@ -1122,9 +1121,9 @@ public final class BCFile {
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)) {
@@ -1132,95 +1131,95 @@ public final class BCFile {
} 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;
-
+
private boolean trackBlocks;
-
+
// 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, boolean trackBlocks) {
this.trackBlocks = trackBlocks;
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) {
if (trackBlocks)
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.
*/
@@ -1229,27 +1228,27 @@ public final class 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.
*/
@@ -1257,37 +1256,37 @@ public final class BCFile {
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;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ef6ba24c/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Chunk.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Chunk.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Chunk.java
deleted file mode 100644
index a075d87..0000000
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Chunk.java
+++ /dev/null
@@ -1,418 +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.accumulo.core.file.rfile.bcfile;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-/**
- * Several related classes to support chunk-encoded sub-streams on top of a regular stream.
- */
-final class Chunk {
-
- /**
- * Prevent the instantiation of class.
- */
- private Chunk() {
- // nothing
- }
-
- /**
- * Decoding a chain of chunks encoded through ChunkEncoder or SingleChunkEncoder.
- */
- static public class ChunkDecoder extends InputStream {
- private DataInputStream in = null;
- private boolean lastChunk;
- private int remain = 0;
- private boolean closed;
-
- public ChunkDecoder() {
- lastChunk = true;
- closed = true;
- }
-
- public void reset(DataInputStream downStream) {
- // no need to wind forward the old input.
- in = downStream;
- lastChunk = false;
- remain = 0;
- closed = false;
- }
-
- /**
- * Constructor
- *
- * @param in
- * The source input stream which contains chunk-encoded data stream.
- */
- public ChunkDecoder(DataInputStream in) {
- this.in = in;
- lastChunk = false;
- closed = false;
- }
-
- /**
- * Have we reached the last chunk.
- *
- * @return true if we have reached the last chunk.
- * @throws java.io.IOException
- */
- public boolean isLastChunk() throws IOException {
- checkEOF();
- return lastChunk;
- }
-
- /**
- * How many bytes remain in the current chunk?
- *
- * @return remaining bytes left in the current chunk.
- * @throws java.io.IOException
- */
- public int getRemain() throws IOException {
- checkEOF();
- return remain;
- }
-
- /**
- * Reading the length of next chunk.
- *
- * @throws java.io.IOException
- * when no more data is available.
- */
- private void readLength() throws IOException {
- remain = Utils.readVInt(in);
- if (remain >= 0) {
- lastChunk = true;
- } else {
- remain = -remain;
- }
- }
-
- /**
- * Check whether we reach the end of the stream.
- *
- * @return false if the chunk encoded stream has more data to read (in which case available() will be greater than 0); true otherwise.
- * @throws java.io.IOException
- * on I/O errors.
- */
- private boolean checkEOF() throws IOException {
- if (isClosed())
- return true;
- while (true) {
- if (remain > 0)
- return false;
- if (lastChunk)
- return true;
- readLength();
- }
- }
-
- @Override
- /*
- * This method never blocks the caller. Returning 0 does not mean we reach the end of the stream.
- */
- public int available() {
- return remain;
- }
-
- @Override
- public int read() throws IOException {
- if (checkEOF())
- return -1;
- int ret = in.read();
- if (ret < 0)
- throw new IOException("Corrupted chunk encoding stream");
- --remain;
- return ret;
- }
-
- @Override
- public int read(byte[] b) throws IOException {
- return read(b, 0, b.length);
- }
-
- @Override
- public int read(byte[] b, int off, int len) throws IOException {
- if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
- throw new IndexOutOfBoundsException();
- }
-
- if (!checkEOF()) {
- int n = Math.min(remain, len);
- int ret = in.read(b, off, n);
- if (ret < 0)
- throw new IOException("Corrupted chunk encoding stream");
- remain -= ret;
- return ret;
- }
- return -1;
- }
-
- @Override
- public long skip(long n) throws IOException {
- if (!checkEOF()) {
- long ret = in.skip(Math.min(remain, n));
- remain -= ret;
- return ret;
- }
- return 0;
- }
-
- @Override
- public boolean markSupported() {
- return false;
- }
-
- public boolean isClosed() {
- return closed;
- }
-
- @Override
- public void close() throws IOException {
- if (closed == false) {
- try {
- while (!checkEOF()) {
- skip(Integer.MAX_VALUE);
- }
- } finally {
- closed = true;
- }
- }
- }
- }
-
- /**
- * Chunk Encoder. Encoding the output data into a chain of chunks in the following sequences: -len1, byte[len1], -len2, byte[len2], ... len_n, byte[len_n].
- * Where len1, len2, ..., len_n are the lengths of the data chunks. Non-terminal chunks have their lengths negated. Non-terminal chunks cannot have length 0.
- * All lengths are in the range of 0 to Integer.MAX_VALUE and are encoded in Utils.VInt format.
- */
- static public class ChunkEncoder extends OutputStream {
- /**
- * The data output stream it connects to.
- */
- private DataOutputStream out;
-
- /**
- * The internal buffer that is only used when we do not know the advertised size.
- */
- private byte buf[];
-
- /**
- * The number of valid bytes in the buffer. This value is always in the range <tt>0</tt> through <tt>buf.length</tt>; elements <tt>buf[0]</tt> through
- * <tt>buf[count-1]</tt> contain valid byte data.
- */
- private int count;
-
- /**
- * Constructor.
- *
- * @param out
- * the underlying output stream.
- * @param buf
- * user-supplied buffer. The buffer would be used exclusively by the ChunkEncoder during its life cycle.
- */
- public ChunkEncoder(DataOutputStream out, byte[] buf) {
- this.out = out;
- this.buf = buf;
- this.count = 0;
- }
-
- /**
- * Write out a chunk.
- *
- * @param chunk
- * The chunk buffer.
- * @param offset
- * Offset to chunk buffer for the beginning of chunk.
- * @param len
- * @param last
- * Is this the last call to flushBuffer?
- */
- private void writeChunk(byte[] chunk, int offset, int len, boolean last) throws IOException {
- if (last) { // always write out the length for the last chunk.
- Utils.writeVInt(out, len);
- if (len > 0) {
- out.write(chunk, offset, len);
- }
- } else {
- if (len > 0) {
- Utils.writeVInt(out, -len);
- out.write(chunk, offset, len);
- }
- }
- }
-
- /**
- * Write out a chunk that is a concatenation of the internal buffer plus user supplied data. This will never be the last block.
- *
- * @param data
- * User supplied data buffer.
- * @param offset
- * Offset to user data buffer.
- * @param len
- * User data buffer size.
- */
- private void writeBufData(byte[] data, int offset, int len) throws IOException {
- if (count + len > 0) {
- Utils.writeVInt(out, -(count + len));
- out.write(buf, 0, count);
- count = 0;
- out.write(data, offset, len);
- }
- }
-
- /**
- * Flush the internal buffer.
- *
- * Is this the last call to flushBuffer?
- *
- * @throws java.io.IOException
- */
- private void flushBuffer() throws IOException {
- if (count > 0) {
- writeChunk(buf, 0, count, false);
- count = 0;
- }
- }
-
- @Override
- public void write(int b) throws IOException {
- if (count >= buf.length) {
- flushBuffer();
- }
- buf[count++] = (byte) b;
- }
-
- @Override
- public void write(byte b[]) throws IOException {
- write(b, 0, b.length);
- }
-
- @Override
- public void write(byte b[], int off, int len) throws IOException {
- if ((len + count) >= buf.length) {
- /*
- * If the input data do not fit in buffer, flush the output buffer and then write the data directly. In this way buffered streams will cascade
- * harmlessly.
- */
- writeBufData(b, off, len);
- return;
- }
-
- System.arraycopy(b, off, buf, count, len);
- count += len;
- }
-
- @Override
- public void flush() throws IOException {
- flushBuffer();
- out.flush();
- }
-
- @Override
- public void close() throws IOException {
- if (buf != null) {
- try {
- writeChunk(buf, 0, count, true);
- } finally {
- buf = null;
- out = null;
- }
- }
- }
- }
-
- /**
- * Encode the whole stream as a single chunk. Expecting to know the size of the chunk up-front.
- */
- static public class SingleChunkEncoder extends OutputStream {
- /**
- * The data output stream it connects to.
- */
- private final DataOutputStream out;
-
- /**
- * The remaining bytes to be written.
- */
- private int remain;
- private boolean closed = false;
-
- /**
- * Constructor.
- *
- * @param out
- * the underlying output stream.
- * @param size
- * The total # of bytes to be written as a single chunk.
- * @throws java.io.IOException
- * if an I/O error occurs.
- */
- public SingleChunkEncoder(DataOutputStream out, int size) throws IOException {
- this.out = out;
- this.remain = size;
- Utils.writeVInt(out, size);
- }
-
- @Override
- public void write(int b) throws IOException {
- if (remain > 0) {
- out.write(b);
- --remain;
- } else {
- throw new IOException("Writing more bytes than advertised size.");
- }
- }
-
- @Override
- public void write(byte b[]) throws IOException {
- write(b, 0, b.length);
- }
-
- @Override
- public void write(byte b[], int off, int len) throws IOException {
- if (remain >= len) {
- out.write(b, off, len);
- remain -= len;
- } else {
- throw new IOException("Writing more bytes than advertised size.");
- }
- }
-
- @Override
- public void flush() throws IOException {
- out.flush();
- }
-
- @Override
- public void close() throws IOException {
- if (closed == true) {
- return;
- }
-
- try {
- if (remain > 0) {
- throw new IOException("Writing less bytes than advertised size.");
- }
- } finally {
- closed = true;
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/ef6ba24c/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java
index e89bb40..66ca07f 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java
@@ -68,11 +68,20 @@ public final class Compression {
}
}
+ /** snappy codec **/
+ public static final String COMPRESSION_SNAPPY = "snappy";
+ /** compression: gzip */
+ public static final String COMPRESSION_GZ = "gz";
+ /** compression: lzo */
+ public static final String COMPRESSION_LZO = "lzo";
+ /** compression: none */
+ public static final String COMPRESSION_NONE = "none";
+
/**
* Compression algorithms.
*/
public static enum Algorithm {
- LZO(TFile.COMPRESSION_LZO) {
+ LZO(COMPRESSION_LZO) {
private transient boolean checked = false;
private static final String defaultClazz = "org.apache.hadoop.io.compress.LzoCodec";
private transient CompressionCodec codec = null;
@@ -137,7 +146,7 @@ public final class Compression {
}
},
- GZ(TFile.COMPRESSION_GZ) {
+ GZ(COMPRESSION_GZ) {
private transient DefaultCodec codec;
@Override
@@ -181,7 +190,7 @@ public final class Compression {
}
},
- NONE(TFile.COMPRESSION_NONE) {
+ NONE(COMPRESSION_NONE) {
@Override
CompressionCodec getCodec() {
return null;
@@ -210,7 +219,7 @@ public final class Compression {
}
},
- SNAPPY(TFile.COMPRESSION_SNAPPY) {
+ SNAPPY(COMPRESSION_SNAPPY) {
// Use base type to avoid compile-time dependencies.
private transient CompressionCodec snappyCodec = null;
private transient boolean checked = false;