You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2016/03/23 02:39:47 UTC
[1/4] hbase git commit: Revert "HBASE-15477 Purge 'next block header'
from cached blocks"
Repository: hbase
Updated Branches:
refs/heads/master 3f3613a23 -> a13d6e000
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
index 69c42c3..66aced0 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
@@ -1317,22 +1317,25 @@ public class BucketCache implements BlockCache, HeapSize {
final AtomicLong realCacheSize) throws CacheFullException, IOException,
BucketAllocatorException {
int len = data.getSerializedLength();
- // This cacheable thing can't be serialized
+ // This cacheable thing can't be serialized...
if (len == 0) return null;
long offset = bucketAllocator.allocateBlock(len);
BucketEntry bucketEntry = new BucketEntry(offset, len, accessCounter, inMemory);
bucketEntry.setDeserialiserReference(data.getDeserializer(), deserialiserMap);
try {
if (data instanceof HFileBlock) {
- // If an instance of HFileBlock, save on some allocations.
- HFileBlock block = (HFileBlock)data;
- ByteBuff sliceBuf = block.getBufferReadOnly();
- ByteBuffer metadata = block.getMetaData();
+ HFileBlock block = (HFileBlock) data;
+ ByteBuff sliceBuf = block.getBufferReadOnlyWithHeader();
+ sliceBuf.rewind();
+ assert len == sliceBuf.limit() + HFileBlock.EXTRA_SERIALIZATION_SPACE ||
+ len == sliceBuf.limit() + block.headerSize() + HFileBlock.EXTRA_SERIALIZATION_SPACE;
+ ByteBuffer extraInfoBuffer = ByteBuffer.allocate(HFileBlock.EXTRA_SERIALIZATION_SPACE);
+ block.serializeExtraInfo(extraInfoBuffer);
if (LOG.isTraceEnabled()) {
LOG.trace("Write offset=" + offset + ", len=" + len);
}
ioEngine.write(sliceBuf, offset);
- ioEngine.write(metadata, offset + len - metadata.limit());
+ ioEngine.write(extraInfoBuffer, offset + len - HFileBlock.EXTRA_SERIALIZATION_SPACE);
} else {
ByteBuffer bb = ByteBuffer.allocate(len);
data.serialize(bb);
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
index ed86a83..e26022e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
@@ -18,7 +18,6 @@
*/
package org.apache.hadoop.hbase.regionserver;
-import java.io.Closeable;
import java.io.IOException;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -33,7 +32,7 @@ import org.apache.hadoop.hbase.client.Scan;
// TODO: Change name from KeyValueScanner to CellScanner only we already have a simple CellScanner
// so this should be something else altogether, a decoration on our base CellScanner. TODO.
// This class shows in CPs so do it all in one swell swoop. HBase-2.0.0.
-public interface KeyValueScanner extends Shipper, Closeable {
+public interface KeyValueScanner extends Shipper {
/**
* The byte array represents for NO_NEXT_INDEXED_KEY;
* The actual value is irrelevant because this is always compared by reference.
@@ -75,7 +74,6 @@ public interface KeyValueScanner extends Shipper, Closeable {
* The default implementation for this would be to return 0. A file having
* lower sequence id will be considered to be the older one.
*/
- // TODO: Implement SequenceId Interface instead.
long getSequenceID();
/**
@@ -139,11 +137,11 @@ public interface KeyValueScanner extends Shipper, Closeable {
* peek KeyValue of scanner has the same row with specified Cell,
* otherwise seek the scanner at the first Cell of the row which is the
* previous row of specified KeyValue
- *
+ *
* @param key seek KeyValue
* @return true if the scanner is at the valid KeyValue, false if such
* KeyValue does not exist
- *
+ *
*/
public boolean backwardSeek(Cell key) throws IOException;
@@ -158,7 +156,7 @@ public interface KeyValueScanner extends Shipper, Closeable {
/**
* Seek the scanner at the first KeyValue of last row
- *
+ *
* @return true if scanner has values left, false if the underlying data is
* empty
* @throws IOException
@@ -171,4 +169,4 @@ public interface KeyValueScanner extends Shipper, Closeable {
* see HFileWriterImpl#getMidpoint, or null if not known.
*/
public Cell getNextIndexedKey();
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
index b6164b2..61eb9b8 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
@@ -1271,7 +1271,7 @@ public class StoreFile {
}
/**
- * @deprecated Do not write further code which depends on this call. Instead
+ * Warning: Do not write further code which depends on this call. Instead
* use getStoreFileScanner() which uses the StoreFileScanner class/interface
* which is the preferred way to scan a store with higher level concepts.
*
@@ -1285,7 +1285,7 @@ public class StoreFile {
}
/**
- * @deprecated Do not write further code which depends on this call. Instead
+ * Warning: Do not write further code which depends on this call. Instead
* use getStoreFileScanner() which uses the StoreFileScanner class/interface
* which is the preferred way to scan a store with higher level concepts.
*
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/CacheTestUtils.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/CacheTestUtils.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/CacheTestUtils.java
index 040685d..69671e2 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/CacheTestUtils.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/CacheTestUtils.java
@@ -40,6 +40,7 @@ import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
import org.apache.hadoop.hbase.nio.ByteBuff;
+import org.apache.hadoop.hbase.nio.SingleByteBuff;
import org.apache.hadoop.hbase.util.ChecksumType;
public class CacheTestUtils {
@@ -65,7 +66,6 @@ public class CacheTestUtils {
/*Post eviction, heapsize should be the same */
assertEquals(heapSize, ((HeapSize) toBeTested).heapSize());
}
-
public static void testCacheMultiThreaded(final BlockCache toBeTested,
final int blockSize, final int numThreads, final int numQueries,
final double passingScore) throws Exception {
@@ -339,16 +339,25 @@ public class CacheTestUtils {
}
- private static HFileBlockPair[] generateHFileBlocks(int blockSize, int numBlocks) {
+ private static HFileBlockPair[] generateHFileBlocks(int blockSize,
+ int numBlocks) {
HFileBlockPair[] returnedBlocks = new HFileBlockPair[numBlocks];
Random rand = new Random();
HashSet<String> usedStrings = new HashSet<String>();
for (int i = 0; i < numBlocks; i++) {
- ByteBuffer cachedBuffer = ByteBuffer.allocate(blockSize);
+
+ // The buffer serialized size needs to match the size of BlockSize. So we
+ // declare our data size to be smaller than it by the serialization space
+ // required.
+
+ SingleByteBuff cachedBuffer = new SingleByteBuff(ByteBuffer.allocate(blockSize
+ - HFileBlock.EXTRA_SERIALIZATION_SPACE));
rand.nextBytes(cachedBuffer.array());
cachedBuffer.rewind();
- int onDiskSizeWithoutHeader = blockSize;
- int uncompressedSizeWithoutHeader = blockSize;
+ int onDiskSizeWithoutHeader = blockSize
+ - HFileBlock.EXTRA_SERIALIZATION_SPACE;
+ int uncompressedSizeWithoutHeader = blockSize
+ - HFileBlock.EXTRA_SERIALIZATION_SPACE;
long prevBlockOffset = rand.nextLong();
BlockType.DATA.write(cachedBuffer);
cachedBuffer.putInt(onDiskSizeWithoutHeader);
@@ -367,7 +376,7 @@ public class CacheTestUtils {
onDiskSizeWithoutHeader, uncompressedSizeWithoutHeader,
prevBlockOffset, cachedBuffer, HFileBlock.DONT_FILL_HEADER,
blockSize,
- onDiskSizeWithoutHeader + HConstants.HFILEBLOCK_HEADER_SIZE, -1, meta);
+ onDiskSizeWithoutHeader + HConstants.HFILEBLOCK_HEADER_SIZE, meta);
String strKey;
/* No conflicting keys */
@@ -386,4 +395,4 @@ public class CacheTestUtils {
BlockCacheKey blockName;
HFileBlock block;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java
index a9d8258..5158e35 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheOnWrite.java
@@ -259,6 +259,7 @@ public class TestCacheOnWrite {
assertTrue(testDescription, scanner.seekTo());
long offset = 0;
+ HFileBlock prevBlock = null;
EnumMap<BlockType, Integer> blockCountByType =
new EnumMap<BlockType, Integer>(BlockType.class);
@@ -266,10 +267,14 @@ public class TestCacheOnWrite {
List<Long> cachedBlocksOffset = new ArrayList<Long>();
Map<Long, HFileBlock> cachedBlocks = new HashMap<Long, HFileBlock>();
while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) {
+ long onDiskSize = -1;
+ if (prevBlock != null) {
+ onDiskSize = prevBlock.getNextBlockOnDiskSizeWithHeader();
+ }
// Flags: don't cache the block, use pread, this is not a compaction.
// Also, pass null for expected block type to avoid checking it.
- HFileBlock block = reader.readBlock(offset, -1, false, true, false, true, null,
- encodingInCache);
+ HFileBlock block = reader.readBlock(offset, onDiskSize, false, true,
+ false, true, null, encodingInCache);
BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(),
offset);
HFileBlock fromCache = (HFileBlock) blockCache.getBlock(blockCacheKey, true, false, true);
@@ -302,6 +307,7 @@ public class TestCacheOnWrite {
assertEquals(
block.getUncompressedSizeWithoutHeader(), fromCache.getUncompressedSizeWithoutHeader());
}
+ prevBlock = block;
offset += block.getOnDiskSizeWithHeader();
BlockType bt = block.getBlockType();
Integer count = blockCountByType.get(bt);
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java
index d91a811..91ab8c0 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java
@@ -94,7 +94,7 @@ public class TestChecksum {
meta = new HFileContextBuilder().withHBaseCheckSum(true).build();
HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl(
is, totalSize, (HFileSystem) fs, path, meta);
- HFileBlock b = hbr.readBlockData(0, -1, false);
+ HFileBlock b = hbr.readBlockData(0, -1, -1, false);
assertEquals(b.getChecksumType(), ChecksumType.getDefaultChecksumType().getCode());
}
@@ -108,14 +108,12 @@ public class TestChecksum {
ChecksumType cktype = itr.next();
Path path = new Path(TEST_UTIL.getDataTestDir(), "checksum" + cktype.getName());
FSDataOutputStream os = fs.create(path);
- HFileContext meta = new HFileContextBuilder().
- withChecksumType(cktype).
- build();
+ HFileContext meta = new HFileContextBuilder()
+ .withChecksumType(cktype).build();
HFileBlock.Writer hbw = new HFileBlock.Writer(null, meta);
DataOutputStream dos = hbw.startWriting(BlockType.DATA);
- for (int i = 0; i < 1000; ++i) {
+ for (int i = 0; i < 1000; ++i)
dos.writeInt(i);
- }
hbw.writeHeaderAndData(os);
int totalSize = hbw.getOnDiskSizeWithHeader();
os.close();
@@ -127,7 +125,7 @@ public class TestChecksum {
meta = new HFileContextBuilder().withHBaseCheckSum(true).build();
HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl(
is, totalSize, (HFileSystem) fs, path, meta);
- HFileBlock b = hbr.readBlockData(0, -1, false);
+ HFileBlock b = hbr.readBlockData(0, -1, -1, false);
ByteBuff data = b.getBufferWithoutHeader();
for (int i = 0; i < 1000; i++) {
assertEquals(i, data.getInt());
@@ -190,7 +188,7 @@ public class TestChecksum {
.withHBaseCheckSum(true)
.build();
HFileBlock.FSReader hbr = new FSReaderImplTest(is, totalSize, fs, path, meta);
- HFileBlock b = hbr.readBlockData(0, -1, pread);
+ HFileBlock b = hbr.readBlockData(0, -1, -1, pread);
b.sanityCheck();
assertEquals(4936, b.getUncompressedSizeWithoutHeader());
assertEquals(algo == GZ ? 2173 : 4936,
@@ -211,17 +209,17 @@ public class TestChecksum {
// requests. Verify that this is correct.
for (int i = 0; i <
HFileBlock.CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD + 1; i++) {
- b = hbr.readBlockData(0, -1, pread);
+ b = hbr.readBlockData(0, -1, -1, pread);
assertEquals(0, HFile.getChecksumFailuresCount());
}
// The next read should have hbase checksum verification reanabled,
// we verify this by assertng that there was a hbase-checksum failure.
- b = hbr.readBlockData(0, -1, pread);
+ b = hbr.readBlockData(0, -1, -1, pread);
assertEquals(1, HFile.getChecksumFailuresCount());
// Since the above encountered a checksum failure, we switch
// back to not checking hbase checksums.
- b = hbr.readBlockData(0, -1, pread);
+ b = hbr.readBlockData(0, -1, -1, pread);
assertEquals(0, HFile.getChecksumFailuresCount());
is.close();
@@ -232,7 +230,7 @@ public class TestChecksum {
assertEquals(false, newfs.useHBaseChecksum());
is = new FSDataInputStreamWrapper(newfs, path);
hbr = new FSReaderImplTest(is, totalSize, newfs, path, meta);
- b = hbr.readBlockData(0, -1, pread);
+ b = hbr.readBlockData(0, -1, -1, pread);
is.close();
b.sanityCheck();
b = b.unpack(meta, hbr);
@@ -316,7 +314,7 @@ public class TestChecksum {
.build();
HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl(new FSDataInputStreamWrapper(
is, nochecksum), totalSize, hfs, path, meta);
- HFileBlock b = hbr.readBlockData(0, -1, pread);
+ HFileBlock b = hbr.readBlockData(0, -1, -1, pread);
is.close();
b.sanityCheck();
assertEquals(dataSize, b.getUncompressedSizeWithoutHeader());
@@ -356,4 +354,5 @@ public class TestChecksum {
return false; // checksum validation failure
}
}
-}
\ No newline at end of file
+}
+
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlock.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlock.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlock.java
index eb87a0c..6748efc 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlock.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlock.java
@@ -320,7 +320,7 @@ public class TestHFileBlock {
.withIncludesTags(includesTag)
.withCompression(algo).build();
HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl(is, totalSize, meta);
- HFileBlock b = hbr.readBlockData(0, -1, pread);
+ HFileBlock b = hbr.readBlockData(0, -1, -1, pread);
is.close();
assertEquals(0, HFile.getChecksumFailuresCount());
@@ -334,15 +334,17 @@ public class TestHFileBlock {
is = fs.open(path);
hbr = new HFileBlock.FSReaderImpl(is, totalSize, meta);
b = hbr.readBlockData(0, 2173 + HConstants.HFILEBLOCK_HEADER_SIZE +
- b.totalChecksumBytes(), pread);
+ b.totalChecksumBytes(), -1, pread);
assertEquals(expected, b);
int wrongCompressedSize = 2172;
try {
b = hbr.readBlockData(0, wrongCompressedSize
- + HConstants.HFILEBLOCK_HEADER_SIZE, pread);
+ + HConstants.HFILEBLOCK_HEADER_SIZE, -1, pread);
fail("Exception expected");
} catch (IOException ex) {
- String expectedPrefix = "Passed in onDiskSizeWithHeader=";
+ String expectedPrefix = "On-disk size without header provided is "
+ + wrongCompressedSize + ", but block header contains "
+ + b.getOnDiskSizeWithoutHeader() + ".";
assertTrue("Invalid exception message: '" + ex.getMessage()
+ "'.\nMessage is expected to start with: '" + expectedPrefix
+ "'", ex.getMessage().startsWith(expectedPrefix));
@@ -422,7 +424,7 @@ public class TestHFileBlock {
HFileBlock blockFromHFile, blockUnpacked;
int pos = 0;
for (int blockId = 0; blockId < numBlocks; ++blockId) {
- blockFromHFile = hbr.readBlockData(pos, -1, pread);
+ blockFromHFile = hbr.readBlockData(pos, -1, -1, pread);
assertEquals(0, HFile.getChecksumFailuresCount());
blockFromHFile.sanityCheck();
pos += blockFromHFile.getOnDiskSizeWithHeader();
@@ -558,7 +560,7 @@ public class TestHFileBlock {
if (detailedLogging) {
LOG.info("Reading block #" + i + " at offset " + curOffset);
}
- HFileBlock b = hbr.readBlockData(curOffset, -1, pread);
+ HFileBlock b = hbr.readBlockData(curOffset, -1, -1, pread);
if (detailedLogging) {
LOG.info("Block #" + i + ": " + b);
}
@@ -572,7 +574,8 @@ public class TestHFileBlock {
// Now re-load this block knowing the on-disk size. This tests a
// different branch in the loader.
- HFileBlock b2 = hbr.readBlockData(curOffset, b.getOnDiskSizeWithHeader(), pread);
+ HFileBlock b2 = hbr.readBlockData(curOffset,
+ b.getOnDiskSizeWithHeader(), -1, pread);
b2.sanityCheck();
assertEquals(b.getBlockType(), b2.getBlockType());
@@ -598,7 +601,7 @@ public class TestHFileBlock {
b = b.unpack(meta, hbr);
// b's buffer has header + data + checksum while
// expectedContents have header + data only
- ByteBuff bufRead = b.getBufferReadOnly();
+ ByteBuff bufRead = b.getBufferWithHeader();
ByteBuffer bufExpected = expectedContents.get(i);
boolean bytesAreCorrect = Bytes.compareTo(bufRead.array(),
bufRead.arrayOffset(),
@@ -681,7 +684,7 @@ public class TestHFileBlock {
HFileBlock b;
try {
long onDiskSizeArg = withOnDiskSize ? expectedSize : -1;
- b = hbr.readBlockData(offset, onDiskSizeArg, pread);
+ b = hbr.readBlockData(offset, onDiskSizeArg, -1, pread);
} catch (IOException ex) {
LOG.error("Error in client " + clientId + " trying to read block at "
+ offset + ", pread=" + pread + ", withOnDiskSize=" +
@@ -716,7 +719,8 @@ public class TestHFileBlock {
protected void testConcurrentReadingInternals() throws IOException,
InterruptedException, ExecutionException {
for (Compression.Algorithm compressAlgo : COMPRESSION_ALGORITHMS) {
- Path path = new Path(TEST_UTIL.getDataTestDir(), "concurrent_reading");
+ Path path =
+ new Path(TEST_UTIL.getDataTestDir(), "concurrent_reading");
Random rand = defaultRandom();
List<Long> offsets = new ArrayList<Long>();
List<BlockType> types = new ArrayList<BlockType>();
@@ -839,7 +843,8 @@ public class TestHFileBlock {
.withBytesPerCheckSum(HFile.DEFAULT_BYTES_PER_CHECKSUM)
.withChecksumType(ChecksumType.NULL).build();
HFileBlock block = new HFileBlock(BlockType.DATA, size, size, -1, buf,
- HFileBlock.FILL_HEADER, -1, 0, -1, meta);
+ HFileBlock.FILL_HEADER, -1,
+ 0, meta);
long byteBufferExpectedSize = ClassSize.align(ClassSize.estimateBase(
new MultiByteBuff(buf).getClass(), true)
+ HConstants.HFILEBLOCK_HEADER_SIZE + size);
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockCompatibility.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockCompatibility.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockCompatibility.java
new file mode 100644
index 0000000..16607b9
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockCompatibility.java
@@ -0,0 +1,750 @@
+/*
+ *
+ * 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.hbase.io.hfile;
+
+import static org.apache.hadoop.hbase.io.compress.Compression.Algorithm.GZ;
+import static org.apache.hadoop.hbase.io.compress.Compression.Algorithm.NONE;
+import static org.junit.Assert.*;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueUtil;
+import org.apache.hadoop.hbase.fs.HFileSystem;
+import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
+import org.apache.hadoop.hbase.io.encoding.HFileBlockDefaultEncodingContext;
+import org.apache.hadoop.hbase.io.encoding.HFileBlockEncodingContext;
+import org.apache.hadoop.hbase.io.hfile.HFileBlock.BlockWritable;
+import org.apache.hadoop.hbase.nio.ByteBuff;
+import org.apache.hadoop.hbase.nio.SingleByteBuff;
+import org.apache.hadoop.hbase.testclassification.IOTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.ChecksumType;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.io.compress.Compressor;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * This class has unit tests to prove that older versions of
+ * HFiles (without checksums) are compatible with current readers.
+ */
+@Category({IOTests.class, SmallTests.class})
+@RunWith(Parameterized.class)
+public class TestHFileBlockCompatibility {
+
+ private static final Log LOG = LogFactory.getLog(TestHFileBlockCompatibility.class);
+ private static final Compression.Algorithm[] COMPRESSION_ALGORITHMS = {
+ NONE, GZ };
+
+ private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+ private HFileSystem fs;
+
+ private final boolean includesMemstoreTS;
+ private final boolean includesTag;
+
+ public TestHFileBlockCompatibility(boolean includesMemstoreTS, boolean includesTag) {
+ this.includesMemstoreTS = includesMemstoreTS;
+ this.includesTag = includesTag;
+ }
+
+ @Parameters
+ public static Collection<Object[]> parameters() {
+ return HBaseTestingUtility.MEMSTORETS_TAGS_PARAMETRIZED;
+ }
+
+ @Before
+ public void setUp() throws IOException {
+ fs = (HFileSystem)HFileSystem.get(TEST_UTIL.getConfiguration());
+ }
+
+ public byte[] createTestV1Block(Compression.Algorithm algo)
+ throws IOException {
+ Compressor compressor = algo.getCompressor();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ OutputStream os = algo.createCompressionStream(baos, compressor, 0);
+ DataOutputStream dos = new DataOutputStream(os);
+ BlockType.META.write(dos); // Let's make this a meta block.
+ TestHFileBlock.writeTestBlockContents(dos);
+ dos.flush();
+ algo.returnCompressor(compressor);
+ return baos.toByteArray();
+ }
+
+ private Writer createTestV2Block(Compression.Algorithm algo)
+ throws IOException {
+ final BlockType blockType = BlockType.DATA;
+ Writer hbw = new Writer(algo, null,
+ includesMemstoreTS, includesTag);
+ DataOutputStream dos = hbw.startWriting(blockType);
+ TestHFileBlock.writeTestBlockContents(dos);
+ // make sure the block is ready by calling hbw.getHeaderAndData()
+ hbw.getHeaderAndData();
+ assertEquals(1000 * 4, hbw.getUncompressedSizeWithoutHeader());
+ hbw.releaseCompressor();
+ return hbw;
+ }
+
+ private String createTestBlockStr(Compression.Algorithm algo,
+ int correctLength) throws IOException {
+ Writer hbw = createTestV2Block(algo);
+ byte[] testV2Block = hbw.getHeaderAndData();
+ int osOffset = HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM + 9;
+ if (testV2Block.length == correctLength) {
+ // Force-set the "OS" field of the gzip header to 3 (Unix) to avoid
+ // variations across operating systems.
+ // See http://www.gzip.org/zlib/rfc-gzip.html for gzip format.
+ testV2Block[osOffset] = 3;
+ }
+ return Bytes.toStringBinary(testV2Block);
+ }
+
+ @Test
+ public void testNoCompression() throws IOException {
+ assertEquals(4000, createTestV2Block(NONE).getBlockForCaching().
+ getUncompressedSizeWithoutHeader());
+ }
+
+ @Test
+ public void testGzipCompression() throws IOException {
+ final String correctTestBlockStr =
+ "DATABLK*\\x00\\x00\\x00:\\x00\\x00\\x0F\\xA0\\xFF\\xFF\\xFF\\xFF"
+ + "\\xFF\\xFF\\xFF\\xFF"
+ // gzip-compressed block: http://www.gzip.org/zlib/rfc-gzip.html
+ + "\\x1F\\x8B" // gzip magic signature
+ + "\\x08" // Compression method: 8 = "deflate"
+ + "\\x00" // Flags
+ + "\\x00\\x00\\x00\\x00" // mtime
+ + "\\x00" // XFL (extra flags)
+ // OS (0 = FAT filesystems, 3 = Unix). However, this field
+ // sometimes gets set to 0 on Linux and Mac, so we reset it to 3.
+ + "\\x03"
+ + "\\xED\\xC3\\xC1\\x11\\x00 \\x08\\xC00DD\\xDD\\x7Fa"
+ + "\\xD6\\xE8\\xA3\\xB9K\\x84`\\x96Q\\xD3\\xA8\\xDB\\xA8e\\xD4c"
+ + "\\xD46\\xEA5\\xEA3\\xEA7\\xE7\\x00LI\\x5Cs\\xA0\\x0F\\x00\\x00";
+ final int correctGzipBlockLength = 82;
+
+ String returnedStr = createTestBlockStr(GZ, correctGzipBlockLength);
+ assertEquals(correctTestBlockStr, returnedStr);
+ }
+
+ @Test
+ public void testReaderV2() throws IOException {
+ if(includesTag) {
+ TEST_UTIL.getConfiguration().setInt("hfile.format.version", 3);
+ }
+ for (Compression.Algorithm algo : COMPRESSION_ALGORITHMS) {
+ for (boolean pread : new boolean[] { false, true }) {
+ LOG.info("testReaderV2: Compression algorithm: " + algo +
+ ", pread=" + pread);
+ Path path = new Path(TEST_UTIL.getDataTestDir(), "blocks_v2_"
+ + algo);
+ FSDataOutputStream os = fs.create(path);
+ Writer hbw = new Writer(algo, null,
+ includesMemstoreTS, includesTag);
+ long totalSize = 0;
+ for (int blockId = 0; blockId < 2; ++blockId) {
+ DataOutputStream dos = hbw.startWriting(BlockType.DATA);
+ for (int i = 0; i < 1234; ++i)
+ dos.writeInt(i);
+ hbw.writeHeaderAndData(os);
+ totalSize += hbw.getOnDiskSizeWithHeader();
+ }
+ os.close();
+
+ FSDataInputStream is = fs.open(path);
+ HFileContext meta = new HFileContextBuilder()
+ .withHBaseCheckSum(false)
+ .withIncludesMvcc(includesMemstoreTS)
+ .withIncludesTags(includesTag)
+ .withCompression(algo)
+ .build();
+ HFileBlock.FSReader hbr =
+ new HFileBlock.FSReaderImpl(new FSDataInputStreamWrapper(is), totalSize, fs, path, meta);
+ HFileBlock b = hbr.readBlockData(0, -1, -1, pread);
+ is.close();
+
+ b.sanityCheck();
+ assertEquals(4936, b.getUncompressedSizeWithoutHeader());
+ assertEquals(algo == GZ ? 2173 : 4936,
+ b.getOnDiskSizeWithoutHeader() - b.totalChecksumBytes());
+ HFileBlock expected = b;
+
+ if (algo == GZ) {
+ is = fs.open(path);
+ hbr = new HFileBlock.FSReaderImpl(new FSDataInputStreamWrapper(is), totalSize, fs, path,
+ meta);
+ b = hbr.readBlockData(0, 2173 + HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM +
+ b.totalChecksumBytes(), -1, pread);
+ assertEquals(expected, b);
+ int wrongCompressedSize = 2172;
+ try {
+ b = hbr.readBlockData(0, wrongCompressedSize
+ + HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM, -1, pread);
+ fail("Exception expected");
+ } catch (IOException ex) {
+ String expectedPrefix = "On-disk size without header provided is "
+ + wrongCompressedSize + ", but block header contains "
+ + b.getOnDiskSizeWithoutHeader() + ".";
+ assertTrue("Invalid exception message: '" + ex.getMessage()
+ + "'.\nMessage is expected to start with: '" + expectedPrefix
+ + "'", ex.getMessage().startsWith(expectedPrefix));
+ }
+ is.close();
+ }
+ }
+ }
+ }
+
+ /**
+ * Test encoding/decoding data blocks.
+ * @throws IOException a bug or a problem with temporary files.
+ */
+ @Test
+ public void testDataBlockEncoding() throws IOException {
+ if(includesTag) {
+ TEST_UTIL.getConfiguration().setInt("hfile.format.version", 3);
+ }
+ final int numBlocks = 5;
+ for (Compression.Algorithm algo : COMPRESSION_ALGORITHMS) {
+ for (boolean pread : new boolean[] { false, true }) {
+ for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
+ LOG.info("testDataBlockEncoding algo " + algo +
+ " pread = " + pread +
+ " encoding " + encoding);
+ Path path = new Path(TEST_UTIL.getDataTestDir(), "blocks_v2_"
+ + algo + "_" + encoding.toString());
+ FSDataOutputStream os = fs.create(path);
+ HFileDataBlockEncoder dataBlockEncoder = (encoding != DataBlockEncoding.NONE) ?
+ new HFileDataBlockEncoderImpl(encoding) : NoOpDataBlockEncoder.INSTANCE;
+ TestHFileBlockCompatibility.Writer hbw =
+ new TestHFileBlockCompatibility.Writer(algo,
+ dataBlockEncoder, includesMemstoreTS, includesTag);
+ long totalSize = 0;
+ final List<Integer> encodedSizes = new ArrayList<Integer>();
+ final List<ByteBuffer> encodedBlocks = new ArrayList<ByteBuffer>();
+ for (int blockId = 0; blockId < numBlocks; ++blockId) {
+ hbw.startWriting(BlockType.DATA);
+ TestHFileBlock.writeTestKeyValues(hbw, blockId, pread, includesTag);
+ hbw.writeHeaderAndData(os);
+ int headerLen = HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM;
+ byte[] encodedResultWithHeader = hbw.getUncompressedDataWithHeader();
+ final int encodedSize = encodedResultWithHeader.length - headerLen;
+ if (encoding != DataBlockEncoding.NONE) {
+ // We need to account for the two-byte encoding algorithm ID that
+ // comes after the 24-byte block header but before encoded KVs.
+ headerLen += DataBlockEncoding.ID_SIZE;
+ }
+ byte[] encodedDataSection =
+ new byte[encodedResultWithHeader.length - headerLen];
+ System.arraycopy(encodedResultWithHeader, headerLen,
+ encodedDataSection, 0, encodedDataSection.length);
+ final ByteBuffer encodedBuf =
+ ByteBuffer.wrap(encodedDataSection);
+ encodedSizes.add(encodedSize);
+ encodedBlocks.add(encodedBuf);
+ totalSize += hbw.getOnDiskSizeWithHeader();
+ }
+ os.close();
+
+ FSDataInputStream is = fs.open(path);
+ HFileContext meta = new HFileContextBuilder()
+ .withHBaseCheckSum(false)
+ .withIncludesMvcc(includesMemstoreTS)
+ .withIncludesTags(includesTag)
+ .withCompression(algo)
+ .build();
+ HFileBlock.FSReaderImpl hbr = new HFileBlock.FSReaderImpl(new FSDataInputStreamWrapper(is),
+ totalSize, fs, path, meta);
+ hbr.setDataBlockEncoder(dataBlockEncoder);
+ hbr.setIncludesMemstoreTS(includesMemstoreTS);
+
+ HFileBlock b;
+ int pos = 0;
+ for (int blockId = 0; blockId < numBlocks; ++blockId) {
+ b = hbr.readBlockData(pos, -1, -1, pread);
+ b.sanityCheck();
+ if (meta.isCompressedOrEncrypted()) {
+ assertFalse(b.isUnpacked());
+ b = b.unpack(meta, hbr);
+ }
+ pos += b.getOnDiskSizeWithHeader();
+
+ assertEquals((int) encodedSizes.get(blockId),
+ b.getUncompressedSizeWithoutHeader());
+ ByteBuff actualBuffer = b.getBufferWithoutHeader();
+ if (encoding != DataBlockEncoding.NONE) {
+ // We expect a two-byte big-endian encoding id.
+ assertEquals(0, actualBuffer.get(0));
+ assertEquals(encoding.getId(), actualBuffer.get(1));
+ actualBuffer.position(2);
+ actualBuffer = actualBuffer.slice();
+ }
+
+ ByteBuffer expectedBuffer = encodedBlocks.get(blockId);
+ expectedBuffer.rewind();
+
+ // test if content matches, produce nice message
+ TestHFileBlock.assertBuffersEqual(new SingleByteBuff(expectedBuffer), actualBuffer,
+ algo, encoding, pread);
+ }
+ is.close();
+ }
+ }
+ }
+ }
+ /**
+ * This is the version of the HFileBlock.Writer that is used to
+ * create V2 blocks with minor version 0. These blocks do not
+ * have hbase-level checksums. The code is here to test
+ * backward compatibility. The reason we do not inherit from
+ * HFileBlock.Writer is because we never ever want to change the code
+ * in this class but the code in HFileBlock.Writer will continually
+ * evolve.
+ */
+ public static final class Writer extends HFileBlock.Writer {
+
+ // These constants are as they were in minorVersion 0.
+ private static final int HEADER_SIZE = HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM;
+ private static final boolean DONT_FILL_HEADER = HFileBlock.DONT_FILL_HEADER;
+ private static final byte[] DUMMY_HEADER = HFileBlock.DUMMY_HEADER_NO_CHECKSUM;
+
+ private enum State {
+ INIT,
+ WRITING,
+ BLOCK_READY
+ };
+
+ /** Writer state. Used to ensure the correct usage protocol. */
+ private State state = State.INIT;
+
+ /** Compression algorithm for all blocks this instance writes. */
+ private final Compression.Algorithm compressAlgo;
+
+ /** Data block encoder used for data blocks */
+ private final HFileDataBlockEncoder dataBlockEncoder;
+
+ private HFileBlockEncodingContext dataBlockEncodingCtx;
+ /** block encoding context for non-data blocks */
+ private HFileBlockDefaultEncodingContext defaultBlockEncodingCtx;
+
+ /**
+ * The stream we use to accumulate data in uncompressed format for each
+ * block. We reset this stream at the end of each block and reuse it. The
+ * header is written as the first {@link #HEADER_SIZE} bytes into this
+ * stream.
+ */
+ private ByteArrayOutputStream baosInMemory;
+
+ /** Compressor, which is also reused between consecutive blocks. */
+ private Compressor compressor;
+
+ /**
+ * Current block type. Set in {@link #startWriting(BlockType)}. Could be
+ * changed in {@link #encodeDataBlockForDisk()} from {@link BlockType#DATA}
+ * to {@link BlockType#ENCODED_DATA}.
+ */
+ private BlockType blockType;
+
+ /**
+ * A stream that we write uncompressed bytes to, which compresses them and
+ * writes them to {@link #baosInMemory}.
+ */
+ private DataOutputStream userDataStream;
+
+ /**
+ * Bytes to be written to the file system, including the header. Compressed
+ * if compression is turned on.
+ */
+ private byte[] onDiskBytesWithHeader;
+
+ /**
+ * Valid in the READY state. Contains the header and the uncompressed (but
+ * potentially encoded, if this is a data block) bytes, so the length is
+ * {@link #uncompressedSizeWithoutHeader} + {@link org.apache.hadoop.hbase.HConstants#HFILEBLOCK_HEADER_SIZE}.
+ */
+ private byte[] uncompressedBytesWithHeader;
+
+ /**
+ * Current block's start offset in the {@link HFile}. Set in
+ * {@link #writeHeaderAndData(FSDataOutputStream)}.
+ */
+ private long startOffset;
+
+ /**
+ * Offset of previous block by block type. Updated when the next block is
+ * started.
+ */
+ private long[] prevOffsetByType;
+
+ /** The offset of the previous block of the same type */
+ private long prevOffset;
+
+ private int unencodedDataSizeWritten;
+
+ public Writer(Compression.Algorithm compressionAlgorithm,
+ HFileDataBlockEncoder dataBlockEncoder, boolean includesMemstoreTS, boolean includesTag) {
+ this(dataBlockEncoder, new HFileContextBuilder().withHBaseCheckSum(false)
+ .withIncludesMvcc(includesMemstoreTS).withIncludesTags(includesTag)
+ .withCompression(compressionAlgorithm).build());
+ }
+
+ public Writer(HFileDataBlockEncoder dataBlockEncoder, HFileContext meta) {
+ super(dataBlockEncoder, meta);
+ compressAlgo = meta.getCompression() == null ? NONE : meta.getCompression();
+ this.dataBlockEncoder = dataBlockEncoder != null ? dataBlockEncoder
+ : NoOpDataBlockEncoder.INSTANCE;
+ defaultBlockEncodingCtx = new HFileBlockDefaultEncodingContext(null, DUMMY_HEADER, meta);
+ dataBlockEncodingCtx = this.dataBlockEncoder.newDataBlockEncodingContext(DUMMY_HEADER, meta);
+ baosInMemory = new ByteArrayOutputStream();
+
+ prevOffsetByType = new long[BlockType.values().length];
+ for (int i = 0; i < prevOffsetByType.length; ++i)
+ prevOffsetByType[i] = -1;
+ }
+
+ /**
+ * Starts writing into the block. The previous block's data is discarded.
+ *
+ * @return the stream the user can write their data into
+ * @throws IOException
+ */
+ public DataOutputStream startWriting(BlockType newBlockType)
+ throws IOException {
+ if (state == State.BLOCK_READY && startOffset != -1) {
+ // We had a previous block that was written to a stream at a specific
+ // offset. Save that offset as the last offset of a block of that type.
+ prevOffsetByType[blockType.getId()] = startOffset;
+ }
+
+ startOffset = -1;
+ blockType = newBlockType;
+
+ baosInMemory.reset();
+ baosInMemory.write(DUMMY_HEADER);
+
+ state = State.WRITING;
+
+ // We will compress it later in finishBlock()
+ userDataStream = new DataOutputStream(baosInMemory);
+ if (newBlockType == BlockType.DATA) {
+ this.dataBlockEncoder.startBlockEncoding(dataBlockEncodingCtx, userDataStream);
+ }
+ this.unencodedDataSizeWritten = 0;
+ return userDataStream;
+ }
+
+ @Override
+ public void write(Cell c) throws IOException {
+ KeyValue kv = KeyValueUtil.ensureKeyValue(c);
+ expectState(State.WRITING);
+ this.dataBlockEncoder.encode(kv, dataBlockEncodingCtx, this.userDataStream);
+ this.unencodedDataSizeWritten += kv.getLength();
+ if (dataBlockEncodingCtx.getHFileContext().isIncludesMvcc()) {
+ this.unencodedDataSizeWritten += WritableUtils.getVIntSize(kv.getSequenceId());
+ }
+ }
+
+ /**
+ * Returns the stream for the user to write to. The block writer takes care
+ * of handling compression and buffering for caching on write. Can only be
+ * called in the "writing" state.
+ *
+ * @return the data output stream for the user to write to
+ */
+ DataOutputStream getUserDataStream() {
+ expectState(State.WRITING);
+ return userDataStream;
+ }
+
+ /**
+ * Transitions the block writer from the "writing" state to the "block
+ * ready" state. Does nothing if a block is already finished.
+ */
+ void ensureBlockReady() throws IOException {
+ Preconditions.checkState(state != State.INIT,
+ "Unexpected state: " + state);
+
+ if (state == State.BLOCK_READY)
+ return;
+
+ // This will set state to BLOCK_READY.
+ finishBlock();
+ }
+
+ /**
+ * An internal method that flushes the compressing stream (if using
+ * compression), serializes the header, and takes care of the separate
+ * uncompressed stream for caching on write, if applicable. Sets block
+ * write state to "block ready".
+ */
+ void finishBlock() throws IOException {
+ if (blockType == BlockType.DATA) {
+ this.dataBlockEncoder.endBlockEncoding(dataBlockEncodingCtx, userDataStream,
+ baosInMemory.toByteArray(), blockType);
+ blockType = dataBlockEncodingCtx.getBlockType();
+ }
+ userDataStream.flush();
+ // This does an array copy, so it is safe to cache this byte array.
+ uncompressedBytesWithHeader = baosInMemory.toByteArray();
+ prevOffset = prevOffsetByType[blockType.getId()];
+
+ // We need to set state before we can package the block up for
+ // cache-on-write. In a way, the block is ready, but not yet encoded or
+ // compressed.
+ state = State.BLOCK_READY;
+ if (blockType == BlockType.DATA || blockType == BlockType.ENCODED_DATA) {
+ onDiskBytesWithHeader = dataBlockEncodingCtx
+ .compressAndEncrypt(uncompressedBytesWithHeader);
+ } else {
+ onDiskBytesWithHeader = defaultBlockEncodingCtx
+ .compressAndEncrypt(uncompressedBytesWithHeader);
+ }
+
+ // put the header for on disk bytes
+ putHeader(onDiskBytesWithHeader, 0,
+ onDiskBytesWithHeader.length,
+ uncompressedBytesWithHeader.length);
+ //set the header for the uncompressed bytes (for cache-on-write)
+ putHeader(uncompressedBytesWithHeader, 0,
+ onDiskBytesWithHeader.length,
+ uncompressedBytesWithHeader.length);
+ }
+
+ /**
+ * Put the header into the given byte array at the given offset.
+ * @param onDiskSize size of the block on disk
+ * @param uncompressedSize size of the block after decompression (but
+ * before optional data block decoding)
+ */
+ private void putHeader(byte[] dest, int offset, int onDiskSize,
+ int uncompressedSize) {
+ offset = blockType.put(dest, offset);
+ offset = Bytes.putInt(dest, offset, onDiskSize - HEADER_SIZE);
+ offset = Bytes.putInt(dest, offset, uncompressedSize - HEADER_SIZE);
+ Bytes.putLong(dest, offset, prevOffset);
+ }
+
+ /**
+ * Similar to {@link #writeHeaderAndData(FSDataOutputStream)}, but records
+ * the offset of this block so that it can be referenced in the next block
+ * of the same type.
+ *
+ * @param out
+ * @throws IOException
+ */
+ public void writeHeaderAndData(FSDataOutputStream out) throws IOException {
+ long offset = out.getPos();
+ if (startOffset != -1 && offset != startOffset) {
+ throw new IOException("A " + blockType + " block written to a "
+ + "stream twice, first at offset " + startOffset + ", then at "
+ + offset);
+ }
+ startOffset = offset;
+
+ writeHeaderAndData((DataOutputStream) out);
+ }
+
+ /**
+ * Writes the header and the compressed data of this block (or uncompressed
+ * data when not using compression) into the given stream. Can be called in
+ * the "writing" state or in the "block ready" state. If called in the
+ * "writing" state, transitions the writer to the "block ready" state.
+ *
+ * @param out the output stream to write the
+ * @throws IOException
+ */
+ private void writeHeaderAndData(DataOutputStream out) throws IOException {
+ ensureBlockReady();
+ out.write(onDiskBytesWithHeader);
+ }
+
+ /**
+ * Returns the header or the compressed data (or uncompressed data when not
+ * using compression) as a byte array. Can be called in the "writing" state
+ * or in the "block ready" state. If called in the "writing" state,
+ * transitions the writer to the "block ready" state.
+ *
+ * @return header and data as they would be stored on disk in a byte array
+ * @throws IOException
+ */
+ public byte[] getHeaderAndData() throws IOException {
+ ensureBlockReady();
+ return onDiskBytesWithHeader;
+ }
+
+ /**
+ * Releases the compressor this writer uses to compress blocks into the
+ * compressor pool. Needs to be called before the writer is discarded.
+ */
+ public void releaseCompressor() {
+ if (compressor != null) {
+ compressAlgo.returnCompressor(compressor);
+ compressor = null;
+ }
+ }
+
+ /**
+ * Returns the on-disk size of the data portion of the block. This is the
+ * compressed size if compression is enabled. Can only be called in the
+ * "block ready" state. Header is not compressed, and its size is not
+ * included in the return value.
+ *
+ * @return the on-disk size of the block, not including the header.
+ */
+ public int getOnDiskSizeWithoutHeader() {
+ expectState(State.BLOCK_READY);
+ return onDiskBytesWithHeader.length - HEADER_SIZE;
+ }
+
+ /**
+ * Returns the on-disk size of the block. Can only be called in the
+ * "block ready" state.
+ *
+ * @return the on-disk size of the block ready to be written, including the
+ * header size
+ */
+ public int getOnDiskSizeWithHeader() {
+ expectState(State.BLOCK_READY);
+ return onDiskBytesWithHeader.length;
+ }
+
+ /**
+ * The uncompressed size of the block data. Does not include header size.
+ */
+ public int getUncompressedSizeWithoutHeader() {
+ expectState(State.BLOCK_READY);
+ return uncompressedBytesWithHeader.length - HEADER_SIZE;
+ }
+
+ /**
+ * The uncompressed size of the block data, including header size.
+ */
+ public int getUncompressedSizeWithHeader() {
+ expectState(State.BLOCK_READY);
+ return uncompressedBytesWithHeader.length;
+ }
+
+ /** @return true if a block is being written */
+ public boolean isWriting() {
+ return state == State.WRITING;
+ }
+
+ /**
+ * Returns the number of bytes written into the current block so far, or
+ * zero if not writing the block at the moment. Note that this will return
+ * zero in the "block ready" state as well.
+ *
+ * @return the number of bytes written
+ */
+ public int blockSizeWritten() {
+ if (state != State.WRITING)
+ return 0;
+ return this.unencodedDataSizeWritten;
+ }
+
+ /**
+ * Returns the header followed by the uncompressed data, even if using
+ * compression. This is needed for storing uncompressed blocks in the block
+ * cache. Can be called in the "writing" state or the "block ready" state.
+ *
+ * @return uncompressed block bytes for caching on write
+ */
+ private byte[] getUncompressedDataWithHeader() {
+ expectState(State.BLOCK_READY);
+
+ return uncompressedBytesWithHeader;
+ }
+
+ private void expectState(State expectedState) {
+ if (state != expectedState) {
+ throw new IllegalStateException("Expected state: " + expectedState +
+ ", actual state: " + state);
+ }
+ }
+
+ /**
+ * Similar to {@link #getUncompressedBufferWithHeader()} but returns a byte
+ * buffer.
+ *
+ * @return uncompressed block for caching on write in the form of a buffer
+ */
+ public ByteBuffer getUncompressedBufferWithHeader() {
+ byte[] b = getUncompressedDataWithHeader();
+ return ByteBuffer.wrap(b, 0, b.length);
+ }
+
+ /**
+ * Takes the given {@link BlockWritable} instance, creates a new block of
+ * its appropriate type, writes the writable into this block, and flushes
+ * the block into the output stream. The writer is instructed not to buffer
+ * uncompressed bytes for cache-on-write.
+ *
+ * @param bw the block-writable object to write as a block
+ * @param out the file system output stream
+ * @throws IOException
+ */
+ public void writeBlock(BlockWritable bw, FSDataOutputStream out)
+ throws IOException {
+ bw.writeToBlock(startWriting(bw.getBlockType()));
+ writeHeaderAndData(out);
+ }
+
+ /**
+ * Creates a new HFileBlock.
+ */
+ public HFileBlock getBlockForCaching() {
+ HFileContext meta = new HFileContextBuilder()
+ .withHBaseCheckSum(false)
+ .withChecksumType(ChecksumType.NULL)
+ .withBytesPerCheckSum(0)
+ .build();
+ return new HFileBlock(blockType, getOnDiskSizeWithoutHeader(),
+ getUncompressedSizeWithoutHeader(), prevOffset,
+ getUncompressedBufferWithHeader(), DONT_FILL_HEADER, startOffset,
+ getOnDiskSizeWithoutHeader(), meta);
+ }
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java
index 470d483..687d3cd 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileBlockIndex.java
@@ -185,7 +185,8 @@ public class TestHFileBlockIndex {
}
missCount += 1;
- prevBlock = realReader.readBlockData(offset, onDiskSize, pread);
+ prevBlock = realReader.readBlockData(offset, onDiskSize,
+ -1, pread);
prevOffset = offset;
prevOnDiskSize = onDiskSize;
prevPread = pread;
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileDataBlockEncoder.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileDataBlockEncoder.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileDataBlockEncoder.java
index 387514e..6f434bb 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileDataBlockEncoder.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileDataBlockEncoder.java
@@ -92,7 +92,8 @@ public class TestHFileDataBlockEncoder {
if (blockEncoder.getDataBlockEncoding() ==
DataBlockEncoding.NONE) {
- assertEquals(block.getBufferReadOnly(), returnedBlock.getBufferReadOnly());
+ assertEquals(block.getBufferWithHeader(),
+ returnedBlock.getBufferWithHeader());
} else {
if (BlockType.ENCODED_DATA != returnedBlock.getBlockType()) {
System.out.println(blockEncoder);
@@ -126,7 +127,7 @@ public class TestHFileDataBlockEncoder {
.build();
HFileBlock block = new HFileBlock(BlockType.DATA, size, size, -1, buf,
HFileBlock.FILL_HEADER, 0,
- 0, -1, hfileContext);
+ 0, hfileContext);
HFileBlock cacheBlock = createBlockOnDisk(kvs, block, useTags);
assertEquals(headerSize, cacheBlock.getDummyHeaderForVersion().length);
}
@@ -197,7 +198,7 @@ public class TestHFileDataBlockEncoder {
.build();
HFileBlock b = new HFileBlock(BlockType.DATA, size, size, -1, buf,
HFileBlock.FILL_HEADER, 0,
- 0, -1, meta);
+ 0, meta);
return b;
}
@@ -219,8 +220,7 @@ public class TestHFileDataBlockEncoder {
byte[] encodedBytes = baos.toByteArray();
size = encodedBytes.length - block.getDummyHeaderForVersion().length;
return new HFileBlock(context.getBlockType(), size, size, -1, ByteBuffer.wrap(encodedBytes),
- HFileBlock.FILL_HEADER, 0, block.getOnDiskDataSizeWithHeader(), -1,
- block.getHFileContext());
+ HFileBlock.FILL_HEADER, 0, block.getOnDiskDataSizeWithHeader(), block.getHFileContext());
}
private void writeBlock(List<Cell> kvs, HFileContext fileContext, boolean useTags)
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileEncryption.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileEncryption.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileEncryption.java
index 3264558..ba3a344 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileEncryption.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileEncryption.java
@@ -99,7 +99,7 @@ public class TestHFileEncryption {
private long readAndVerifyBlock(long pos, HFileContext ctx, HFileBlock.FSReaderImpl hbr, int size)
throws IOException {
- HFileBlock b = hbr.readBlockData(pos, -1, false);
+ HFileBlock b = hbr.readBlockData(pos, -1, -1, false);
assertEquals(0, HFile.getChecksumFailuresCount());
b.sanityCheck();
assertFalse(b.isUnpacked());
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileWriterV3.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileWriterV3.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileWriterV3.java
index 983ec2f..c7eb11b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileWriterV3.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileWriterV3.java
@@ -218,7 +218,7 @@ public class TestHFileWriterV3 {
fsdis.seek(0);
long curBlockPos = 0;
while (curBlockPos <= trailer.getLastDataBlockOffset()) {
- HFileBlock block = blockReader.readBlockData(curBlockPos, -1, false)
+ HFileBlock block = blockReader.readBlockData(curBlockPos, -1, -1, false)
.unpack(context, blockReader);
assertEquals(BlockType.DATA, block.getBlockType());
ByteBuff buf = block.getBufferWithoutHeader();
@@ -279,14 +279,13 @@ public class TestHFileWriterV3 {
while (fsdis.getPos() < trailer.getLoadOnOpenDataOffset()) {
LOG.info("Current offset: " + fsdis.getPos() + ", scanning until " +
trailer.getLoadOnOpenDataOffset());
- HFileBlock block = blockReader.readBlockData(curBlockPos, -1, false)
+ HFileBlock block = blockReader.readBlockData(curBlockPos, -1, -1, false)
.unpack(context, blockReader);
assertEquals(BlockType.META, block.getBlockType());
Text t = new Text();
ByteBuff buf = block.getBufferWithoutHeader();
if (Writables.getWritable(buf.array(), buf.arrayOffset(), buf.limit(), t) == null) {
- throw new IOException("Failed to deserialize block " + this +
- " into a " + t.getClass().getSimpleName());
+ throw new IOException("Failed to deserialize block " + this + " into a " + t.getClass().getSimpleName());
}
Text expectedText =
(metaCounter == 0 ? new Text("Paris") : metaCounter == 1 ? new Text(
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java
index d20ba2b..69a77bf 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestPrefetch.java
@@ -78,8 +78,14 @@ public class TestPrefetch {
// Check that all of the data blocks were preloaded
BlockCache blockCache = cacheConf.getBlockCache();
long offset = 0;
+ HFileBlock prevBlock = null;
while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) {
- HFileBlock block = reader.readBlock(offset, -1, false, true, false, true, null, null);
+ long onDiskSize = -1;
+ if (prevBlock != null) {
+ onDiskSize = prevBlock.getNextBlockOnDiskSizeWithHeader();
+ }
+ HFileBlock block = reader.readBlock(offset, onDiskSize, false, true, false, true, null,
+ null);
BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(), offset);
boolean isCached = blockCache.getBlock(blockCacheKey, true, false, true) != null;
if (block.getBlockType() == BlockType.DATA ||
@@ -87,6 +93,7 @@ public class TestPrefetch {
block.getBlockType() == BlockType.INTERMEDIATE_INDEX) {
assertTrue(isCached);
}
+ prevBlock = block;
offset += block.getOnDiskSizeWithHeader();
}
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
index 2357bef..0916fe6 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
@@ -227,10 +227,15 @@ public class TestCacheOnWriteInSchema {
assertTrue(testDescription, scanner.seekTo());
// Cribbed from io.hfile.TestCacheOnWrite
long offset = 0;
+ HFileBlock prevBlock = null;
while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) {
+ long onDiskSize = -1;
+ if (prevBlock != null) {
+ onDiskSize = prevBlock.getNextBlockOnDiskSizeWithHeader();
+ }
// Flags: don't cache the block, use pread, this is not a compaction.
// Also, pass null for expected block type to avoid checking it.
- HFileBlock block = reader.readBlock(offset, -1, false, true,
+ HFileBlock block = reader.readBlock(offset, onDiskSize, false, true,
false, true, null, DataBlockEncoding.NONE);
BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(),
offset);
@@ -244,6 +249,7 @@ public class TestCacheOnWriteInSchema {
"block: " + block + "\n" +
"blockCacheKey: " + blockCacheKey);
}
+ prevBlock = block;
offset += block.getOnDiskSizeWithHeader();
}
} finally {
[3/4] hbase git commit: Revert "HBASE-15477 Purge 'next block header'
from cached blocks" Overcommit. Revert.
Posted by st...@apache.org.
Revert "HBASE-15477 Purge 'next block header' from cached blocks"
Overcommit. Revert.
This reverts commit ef94b552399d75c0c16aca0109a39b66f15af892.
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/c49d0caf
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/c49d0caf
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/c49d0caf
Branch: refs/heads/master
Commit: c49d0caf53b27c0c22481e074c807a5ad4753eda
Parents: 54a543d
Author: stack <st...@apache.org>
Authored: Tue Mar 22 18:38:29 2016 -0700
Committer: stack <st...@apache.org>
Committed: Tue Mar 22 18:38:29 2016 -0700
----------------------------------------------------------------------
.../hbase/regionserver/KeyValueScanner.java.rej | 17 -----------------
1 file changed, 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/c49d0caf/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java.rej
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java.rej b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java.rej
deleted file mode 100644
index 4e3ebd4..0000000
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java.rej
+++ /dev/null
@@ -1,17 +0,0 @@
-diff a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java (rejected hunks)
-@@ -29,7 +30,7 @@ import org.apache.hadoop.hbase.client.Scan;
- * Scanner that returns the next KeyValue.
- */
- @InterfaceAudience.Private
--public interface KeyValueScanner extends Shipper {
-+public interface KeyValueScanner extends Shipper, Closeable {
- /**
- * The byte array represents for NO_NEXT_INDEXED_KEY;
- * The actual value is irrelevant because this is always compared by reference.
-@@ -165,4 +167,4 @@ public interface KeyValueScanner extends Shipper {
- * if known, or null otherwise
- */
- public Cell getNextIndexedKey();
--}
-+}
-\ No newline at end of file
[2/4] hbase git commit: Revert "HBASE-15477 Purge 'next block header'
from cached blocks"
Posted by st...@apache.org.
Revert "HBASE-15477 Purge 'next block header' from cached blocks"
Overcommit. Revert to fix.
This reverts commit 000117ad9fd7eb59074c9bb0da2cf1f9544d4bed.
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/54a543de
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/54a543de
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/54a543de
Branch: refs/heads/master
Commit: 54a543de229b358b438c920f04f7f2d1ff767cab
Parents: 3f3613a
Author: stack <st...@apache.org>
Authored: Tue Mar 22 18:37:25 2016 -0700
Committer: stack <st...@apache.org>
Committed: Tue Mar 22 18:37:25 2016 -0700
----------------------------------------------------------------------
.../apache/hadoop/hbase/io/hfile/BlockType.java | 4 -
.../hbase/io/hfile/HFileContextBuilder.java | 20 -
.../org/apache/hadoop/hbase/nio/ByteBuff.java | 6 -
.../hbase/io/hfile/MemcachedBlockCache.java | 2 +-
.../hadoop/hbase/io/hfile/ChecksumUtil.java | 5 +-
.../hadoop/hbase/io/hfile/HFileBlock.java | 997 ++++++++++---------
.../hadoop/hbase/io/hfile/HFileBlockIndex.java | 2 +-
.../hadoop/hbase/io/hfile/HFileReaderImpl.java | 26 +-
.../hadoop/hbase/io/hfile/HFileScanner.java | 12 -
.../hbase/io/hfile/bucket/BucketCache.java | 15 +-
.../hbase/regionserver/KeyValueScanner.java | 12 +-
.../hadoop/hbase/regionserver/StoreFile.java | 4 +-
.../hadoop/hbase/io/hfile/CacheTestUtils.java | 23 +-
.../hadoop/hbase/io/hfile/TestCacheOnWrite.java | 10 +-
.../hadoop/hbase/io/hfile/TestChecksum.java | 27 +-
.../hadoop/hbase/io/hfile/TestHFileBlock.java | 27 +-
.../io/hfile/TestHFileBlockCompatibility.java | 750 ++++++++++++++
.../hbase/io/hfile/TestHFileBlockIndex.java | 3 +-
.../io/hfile/TestHFileDataBlockEncoder.java | 10 +-
.../hbase/io/hfile/TestHFileEncryption.java | 2 +-
.../hbase/io/hfile/TestHFileWriterV3.java | 7 +-
.../hadoop/hbase/io/hfile/TestPrefetch.java | 9 +-
.../regionserver/TestCacheOnWriteInSchema.java | 8 +-
23 files changed, 1374 insertions(+), 607 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockType.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockType.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockType.java
index 32eb0b2..4228f57 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockType.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockType.java
@@ -132,10 +132,6 @@ public enum BlockType {
out.write(magic);
}
- public void write(ByteBuffer buf) {
- buf.put(magic);
- }
-
public void write(ByteBuff buf) {
buf.put(magic);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileContextBuilder.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileContextBuilder.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileContextBuilder.java
index a6645a6..6d3bb13 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileContextBuilder.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileContextBuilder.java
@@ -55,26 +55,6 @@ public class HFileContextBuilder {
private String hfileName = null;
- public HFileContextBuilder() {}
-
- /**
- * Use this constructor if you want to change a few settings only in another context.
- */
- public HFileContextBuilder(final HFileContext hfc) {
- this.usesHBaseChecksum = hfc.isUseHBaseChecksum();
- this.includesMvcc = hfc.isIncludesMvcc();
- this.includesTags = hfc.isIncludesTags();
- this.compression = hfc.getCompression();
- this.compressTags = hfc.isCompressTags();
- this.checksumType = hfc.getChecksumType();
- this.bytesPerChecksum = hfc.getBytesPerChecksum();
- this.blocksize = hfc.getBlocksize();
- this.encoding = hfc.getDataBlockEncoding();
- this.cryptoContext = hfc.getEncryptionContext();
- this.fileCreateTime = hfc.getFileCreateTime();
- this.hfileName = hfc.getHFileName();
- }
-
public HFileContextBuilder withHBaseCheckSum(boolean useHBaseCheckSum) {
this.usesHBaseChecksum = useHBaseCheckSum;
return this;
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-common/src/main/java/org/apache/hadoop/hbase/nio/ByteBuff.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/nio/ByteBuff.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/nio/ByteBuff.java
index 183a031..1e0e957 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/nio/ByteBuff.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/nio/ByteBuff.java
@@ -496,12 +496,6 @@ public abstract class ByteBuff {
return -(low + 1); // key not found.
}
- @Override
- public String toString() {
- return this.getClass().getSimpleName() + "[pos=" + position() + ", lim=" + limit() +
- ", cap= " + capacity() + "]";
- }
-
public static String toStringBinary(final ByteBuff b, int off, int len) {
StringBuilder result = new StringBuilder();
// Just in case we are passed a 'len' that is > buffer length...
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java
----------------------------------------------------------------------
diff --git a/hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java b/hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java
index ae871c4..536872e 100644
--- a/hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java
+++ b/hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java
@@ -260,7 +260,7 @@ public class MemcachedBlockCache implements BlockCache {
public HFileBlock decode(CachedData d) {
try {
ByteBuff buf = new SingleByteBuff(ByteBuffer.wrap(d.getData()));
- return (HFileBlock) HFileBlock.BLOCK_DESERIALIZER.deserialize(buf, true,
+ return (HFileBlock) HFileBlock.blockDeserializer.deserialize(buf, true,
MemoryType.EXCLUSIVE);
} catch (IOException e) {
LOG.warn("Error deserializing data from memcached",e);
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ChecksumUtil.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ChecksumUtil.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ChecksumUtil.java
index b0b1714..69f4330 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ChecksumUtil.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ChecksumUtil.java
@@ -91,7 +91,7 @@ public class ChecksumUtil {
// If this is an older version of the block that does not have
// checksums, then return false indicating that checksum verification
- // did not succeed. Actually, this method should never be called
+ // did not succeed. Actually, this methiod should never be called
// when the minorVersion is 0, thus this is a defensive check for a
// cannot-happen case. Since this is a cannot-happen case, it is
// better to return false to indicate a checksum validation failure.
@@ -141,7 +141,8 @@ public class ChecksumUtil {
* @return The number of bytes needed to store the checksum values
*/
static long numBytes(long datasize, int bytesPerChecksum) {
- return numChunks(datasize, bytesPerChecksum) * HFileBlock.CHECKSUM_SIZE;
+ return numChunks(datasize, bytesPerChecksum) *
+ HFileBlock.CHECKSUM_SIZE;
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java
index f3402da..6268f2e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java
@@ -56,131 +56,50 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
/**
- * Reads {@link HFile} version 2 blocks to HFiles and via {@link Cacheable} Interface to caches.
- * Version 2 was introduced in hbase-0.92.0. No longer has support for version 1 blocks since
- * hbase-1.3.0.
- *
- * <p>Version 1 was the original file block. Version 2 was introduced when we changed the hbase file
- * format to support multi-level block indexes and compound bloom filters (HBASE-3857).
+ * Reads {@link HFile} version 1 and version 2 blocks but writes version 2 blocks only.
+ * Version 2 was introduced in hbase-0.92.0. Does read and write out to the filesystem but also
+ * the read and write to Cache.
*
+ * <h3>HFileBlock: Version 1</h3>
+ * As of this writing, there should be no more version 1 blocks found out in the wild. Version 2
+ * as introduced in hbase-0.92.0.
+ * In version 1 all blocks are always compressed or uncompressed, as
+ * specified by the {@link HFile}'s compression algorithm, with a type-specific
+ * magic record stored in the beginning of the compressed data (i.e. one needs
+ * to uncompress the compressed block to determine the block type). There is
+ * only a single compression algorithm setting for all blocks. Offset and size
+ * information from the block index are required to read a block.
* <h3>HFileBlock: Version 2</h3>
* In version 2, a block is structured as follows:
* <ul>
- * <li><b>Header:</b> See Writer#putHeader() for where header is written; header total size is
- * HFILEBLOCK_HEADER_SIZE
+ * <li><b>Header:</b> See Writer#putHeader(); header total size is HFILEBLOCK_HEADER_SIZE)
* <ul>
- * <li>0. blockType: Magic record identifying the {@link BlockType} (8 bytes):
- * e.g. <code>DATABLK*</code>
- * <li>1. onDiskSizeWithoutHeader: Compressed -- a.k.a 'on disk' -- block size, excluding header,
- * but including tailing checksum bytes (4 bytes)
- * <li>2. uncompressedSizeWithoutHeader: Uncompressed block size, excluding header, and excluding
- * checksum bytes (4 bytes)
- * <li>3. prevBlockOffset: The offset of the previous block of the same type (8 bytes). This is
+ * <li>Magic record identifying the {@link BlockType} (8 bytes): e.g. <code>DATABLK*</code>
+ * <li>Compressed -- a.k.a 'on disk' -- block size, excluding header, but including
+ * tailing checksum bytes (4 bytes)
+ * <li>Uncompressed block size, excluding header, and excluding checksum bytes (4 bytes)
+ * <li>The offset of the previous block of the same type (8 bytes). This is
* used to navigate to the previous block without having to go to the block index
- * <li>4: For minorVersions >=1, the ordinal describing checksum type (1 byte)
- * <li>5: For minorVersions >=1, the number of data bytes/checksum chunk (4 bytes)
- * <li>6: onDiskDataSizeWithHeader: For minorVersions >=1, the size of data 'on disk', including
- * header, excluding checksums (4 bytes)
+ * <li>For minorVersions >=1, the ordinal describing checksum type (1 byte)
+ * <li>For minorVersions >=1, the number of data bytes/checksum chunk (4 bytes)
+ * <li>For minorVersions >=1, the size of data 'on disk', including header,
+ * excluding checksums (4 bytes)
* </ul>
* </li>
- * <li><b>Raw/Compressed/Encrypted/Encoded data:</b> The compression
- * algorithm is the same for all the blocks in an {@link HFile}. If compression is NONE, this is
- * just raw, serialized Cells.
+ * <li><b>Raw/Compressed/Encrypted/Encoded data:</b> The compression algorithm is the
+ * same for all the blocks in the {@link HFile}, similarly to what was done in
+ * version 1. If compression is NONE, this is just raw, serialized Cells.
* <li><b>Tail:</b> For minorVersions >=1, a series of 4 byte checksums, one each for
* the number of bytes specified by bytesPerChecksum.
* </ul>
- *
- * <h3>Caching</h3>
- * Caches cache whole blocks with trailing checksums if any. We then tag on some metadata, the
- * content of BLOCK_METADATA_SPACE which will be flag on if we are doing 'hbase'
- * checksums and then the offset into the file which is needed when we re-make a cache key
- * when we return the block to the cache as 'done'. See {@link Cacheable#serialize(ByteBuffer)} and
- * {@link Cacheable#getDeserializer()}.
- *
- * <p>TODO: Should we cache the checksums? Down in Writer#getBlockForCaching(CacheConfig) where
- * we make a block to cache-on-write, there is an attempt at turning off checksums. This is not the
- * only place we get blocks to cache. We also will cache the raw return from an hdfs read. In this
- * case, the checksums may be present. If the cache is backed by something that doesn't do ECC,
- * say an SSD, we might want to preserve checksums. For now this is open question.
- * <p>TODO: Over in BucketCache, we save a block allocation by doing a custom serialization.
- * Be sure to change it if serialization changes in here. Could we add a method here that takes an
- * IOEngine and that then serializes to it rather than expose our internals over in BucketCache?
- * IOEngine is in the bucket subpackage. Pull it up? Then this class knows about bucketcache. Ugh.
+ * <p>Be aware that when we read from HDFS, we overread pulling in the next blocks' header too.
+ * We do this to save having to do two seeks to read an HFileBlock; a seek to read the header
+ * to figure lengths, etc., and then another seek to pull in the data.
*/
@InterfaceAudience.Private
public class HFileBlock implements Cacheable {
private static final Log LOG = LogFactory.getLog(HFileBlock.class);
- /** Type of block. Header field 0. */
- private BlockType blockType;
-
- /**
- * Size on disk excluding header, including checksum. Header field 1.
- * @see Writer#putHeader(byte[], int, int, int, int)
- */
- private int onDiskSizeWithoutHeader;
-
- /**
- * Size of pure data. Does not include header or checksums. Header field 2.
- * @see Writer#putHeader(byte[], int, int, int, int)
- */
- private int uncompressedSizeWithoutHeader;
-
- /**
- * The offset of the previous block on disk. Header field 3.
- * @see Writer#putHeader(byte[], int, int, int, int)
- */
- private long prevBlockOffset;
-
- /**
- * Size on disk of header + data. Excludes checksum. Header field 6,
- * OR calculated from {@link #onDiskSizeWithoutHeader} when using HDFS checksum.
- * @see Writer#putHeader(byte[], int, int, int, int)
- */
- private int onDiskDataSizeWithHeader;
-
-
- /**
- * The in-memory representation of the hfile block. Can be on or offheap. Can be backed by
- * a single ByteBuffer or by many. Make no assumptions.
- *
- * <p>Be careful reading from this <code>buf</code>. Duplicate and work on the duplicate or if
- * not, be sure to reset position and limit else trouble down the road.
- *
- * <p>TODO: Make this read-only once made.
- *
- * <p>We are using the ByteBuff type. ByteBuffer is not extensible yet we need to be able to have
- * a ByteBuffer-like API across multiple ByteBuffers reading from a cache such as BucketCache.
- * So, we have this ByteBuff type. Unfortunately, it is spread all about HFileBlock. Would be
- * good if could be confined to cache-use only but hard-to-do.
- */
- private ByteBuff buf;
-
- /** Meta data that holds meta information on the hfileblock.
- */
- private HFileContext fileContext;
-
- /**
- * The offset of this block in the file. Populated by the reader for
- * convenience of access. This offset is not part of the block header.
- */
- private long offset = UNSET;
-
- private MemoryType memType = MemoryType.EXCLUSIVE;
-
- /**
- * The on-disk size of the next block, including the header and checksums if present, obtained by
- * peeking into the first {@link HConstants#HFILEBLOCK_HEADER_SIZE} bytes of the next block's
- * header, or UNSET if unknown.
- *
- * Blocks try to carry the size of the next block to read in this data member. They will even have
- * this value when served from cache. Could save a seek in the case where we are iterating through
- * a file and some of the blocks come from cache. If from cache, then having this info to hand
- * will save us doing a seek to read the header so we can read the body of a block.
- * TODO: see how effective this is at saving seeks.
- */
- private int nextBlockOnDiskSize = UNSET;
-
/**
* On a checksum failure, do these many succeeding read requests using hdfs checksums before
* auto-reenabling hbase checksum verification.
@@ -196,18 +115,14 @@ public class HFileBlock implements Cacheable {
(int)ClassSize.estimateBase(MultiByteBuff.class, false);
/**
- * Space for metadata on a block that gets stored along with the block when we cache it.
- * There are a few bytes stuck on the end of the HFileBlock that we pull in from HDFS (note,
+ * See #blockDeserializer method for more info.
+ * 13 bytes of extra stuff stuck on the end of the HFileBlock that we pull in from HDFS (note,
* when we read from HDFS, we pull in an HFileBlock AND the header of the next block if one).
- * 8 bytes are offset of this block (long) in the file. Offset is important because
- * used when we remake the CacheKey when we return the block to cache when done. There is also
- * a flag on whether checksumming is being done by hbase or not. See class comment for note on
- * uncertain state of checksumming of blocks that come out of cache (should we or should we not?).
- * Finally there 4 bytes to hold the length of the next block which can save a seek on occasion.
- * <p>This EXTRA came in with original commit of the bucketcache, HBASE-7404. Was formerly
- * known as EXTRA_SERIALIZATION_SPACE.
+ * The 13 bytes are: usesHBaseChecksum (1 byte) + offset of this block (long) +
+ * nextBlockOnDiskSizeWithHeader (int).
*/
- static final int BLOCK_METADATA_SPACE = Bytes.SIZEOF_BYTE + Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
+ public static final int EXTRA_SERIALIZATION_SPACE =
+ Bytes.SIZEOF_BYTE + Bytes.SIZEOF_INT + Bytes.SIZEOF_LONG;
/**
* Each checksum value is an integer that can be stored in 4 bytes.
@@ -220,47 +135,57 @@ public class HFileBlock implements Cacheable {
/**
* Used deserializing blocks from Cache.
*
- * <code>
+ * Serializing to cache is a little hard to follow. See Writer#finishBlock for where it is done.
+ * When we start to append to a new HFileBlock,
+ * we skip over where the header should go before we start adding Cells. When the block is
+ * done, we'll then go back and fill in the header and the checksum tail. Be aware that what
+ * gets serialized into the blockcache is a byte array that contains an HFileBlock followed by
+ * its checksums and then the header of the next HFileBlock (needed to help navigate), followed
+ * again by an extra 13 bytes of meta info needed when time to recreate the HFileBlock from cache.
+ *
* ++++++++++++++
* + HFileBlock +
* ++++++++++++++
- * + Checksums + <= Optional
+ * + Checksums +
+ * ++++++++++++++
+ * + NextHeader +
* ++++++++++++++
- * + Metadata! +
+ * + ExtraMeta! +
* ++++++++++++++
- * </code>
- * @see #serialize(ByteBuffer)
+ *
+ * TODO: Fix it so we do NOT put the NextHeader into blockcache. It is not necessary.
*/
- static final CacheableDeserializer<Cacheable> BLOCK_DESERIALIZER =
+ static final CacheableDeserializer<Cacheable> blockDeserializer =
new CacheableDeserializer<Cacheable>() {
public HFileBlock deserialize(ByteBuff buf, boolean reuse, MemoryType memType)
throws IOException {
- // The buf has the file block followed by block metadata.
- // Set limit to just before the BLOCK_METADATA_SPACE then rewind.
- buf.limit(buf.limit() - BLOCK_METADATA_SPACE).rewind();
- // Get a new buffer to pass the HFileBlock for it to 'own'.
- ByteBuff newByteBuff;
+ // Rewind to just before the EXTRA_SERIALIZATION_SPACE.
+ buf.limit(buf.limit() - HFileBlock.EXTRA_SERIALIZATION_SPACE).rewind();
+ // Get a new buffer to pass the deserialized HFileBlock for it to 'own'.
+ ByteBuff newByteBuffer;
if (reuse) {
- newByteBuff = buf.slice();
+ newByteBuffer = buf.slice();
} else {
int len = buf.limit();
- newByteBuff = new SingleByteBuff(ByteBuffer.allocate(len));
- newByteBuff.put(0, buf, buf.position(), len);
+ newByteBuffer = new SingleByteBuff(ByteBuffer.allocate(len));
+ newByteBuffer.put(0, buf, buf.position(), len);
}
- // Read out the BLOCK_METADATA_SPACE content and shove into our HFileBlock.
+ // Read out the EXTRA_SERIALIZATION_SPACE content and shove into our HFileBlock.
buf.position(buf.limit());
- buf.limit(buf.limit() + HFileBlock.BLOCK_METADATA_SPACE);
+ buf.limit(buf.limit() + HFileBlock.EXTRA_SERIALIZATION_SPACE);
boolean usesChecksum = buf.get() == (byte)1;
- long offset = buf.getLong();
- int nextBlockOnDiskSize = buf.getInt();
- HFileBlock hFileBlock =
- new HFileBlock(newByteBuff, usesChecksum, memType, offset, nextBlockOnDiskSize, null);
+ HFileBlock hFileBlock = new HFileBlock(newByteBuffer, usesChecksum, memType);
+ hFileBlock.offset = buf.getLong();
+ hFileBlock.nextBlockOnDiskSizeWithHeader = buf.getInt();
+ if (hFileBlock.hasNextBlockHeader()) {
+ hFileBlock.buf.limit(hFileBlock.buf.limit() - hFileBlock.headerSize());
+ }
return hFileBlock;
}
@Override
public int getDeserialiserIdentifier() {
- return DESERIALIZER_IDENTIFIER;
+ return deserializerIdentifier;
}
@Override
@@ -270,36 +195,65 @@ public class HFileBlock implements Cacheable {
}
};
- private static final int DESERIALIZER_IDENTIFIER;
+ private static final int deserializerIdentifier;
static {
- DESERIALIZER_IDENTIFIER =
- CacheableDeserializerIdManager.registerDeserializer(BLOCK_DESERIALIZER);
+ deserializerIdentifier = CacheableDeserializerIdManager
+ .registerDeserializer(blockDeserializer);
}
+ /** Type of block. Header field 0. */
+ private BlockType blockType;
+
/**
- * Copy constructor. Creates a shallow copy of {@code that}'s buffer.
+ * Size on disk excluding header, including checksum. Header field 1.
+ * @see Writer#putHeader(byte[], int, int, int, int)
*/
- private HFileBlock(HFileBlock that) {
- this.blockType = that.blockType;
- this.onDiskSizeWithoutHeader = that.onDiskSizeWithoutHeader;
- this.uncompressedSizeWithoutHeader = that.uncompressedSizeWithoutHeader;
- this.prevBlockOffset = that.prevBlockOffset;
- this.buf = that.buf.duplicate();
- this.offset = that.offset;
- this.onDiskDataSizeWithHeader = that.onDiskDataSizeWithHeader;
- this.fileContext = that.fileContext;
- this.nextBlockOnDiskSize = that.nextBlockOnDiskSize;
- }
+ private int onDiskSizeWithoutHeader;
+
+ /**
+ * Size of pure data. Does not include header or checksums. Header field 2.
+ * @see Writer#putHeader(byte[], int, int, int, int)
+ */
+ private final int uncompressedSizeWithoutHeader;
+
+ /**
+ * The offset of the previous block on disk. Header field 3.
+ * @see Writer#putHeader(byte[], int, int, int, int)
+ */
+ private final long prevBlockOffset;
+
+ /**
+ * Size on disk of header + data. Excludes checksum. Header field 6,
+ * OR calculated from {@link #onDiskSizeWithoutHeader} when using HDFS checksum.
+ * @see Writer#putHeader(byte[], int, int, int, int)
+ */
+ private final int onDiskDataSizeWithHeader;
+
+ /** The in-memory representation of the hfile block */
+ private ByteBuff buf;
+
+ /** Meta data that holds meta information on the hfileblock */
+ private HFileContext fileContext;
+
+ /**
+ * The offset of this block in the file. Populated by the reader for
+ * convenience of access. This offset is not part of the block header.
+ */
+ private long offset = UNSET;
+
+ /**
+ * The on-disk size of the next block, including the header, obtained by
+ * peeking into the first {@link HConstants#HFILEBLOCK_HEADER_SIZE} bytes of the next block's
+ * header, or -1 if unknown.
+ */
+ private int nextBlockOnDiskSizeWithHeader = UNSET;
+
+ private MemoryType memType = MemoryType.EXCLUSIVE;
/**
* Creates a new {@link HFile} block from the given fields. This constructor
* is used when the block data has already been read and uncompressed,
- * and is sitting in a byte buffer and we want to stuff the block into cache.
- * See {@link Writer#getBlockForCaching(CacheConfig)}.
- *
- * <p>TODO: The caller presumes no checksumming
- * required of this block instance since going into cache; checksum already verified on
- * underlying block data pulled in from filesystem. Is that correct? What if cache is SSD?
+ * and is sitting in a byte buffer.
*
* @param blockType the type of this block, see {@link BlockType}
* @param onDiskSizeWithoutHeader see {@link #onDiskSizeWithoutHeader}
@@ -313,94 +267,86 @@ public class HFileBlock implements Cacheable {
* @param fileContext HFile meta data
*/
HFileBlock(BlockType blockType, int onDiskSizeWithoutHeader, int uncompressedSizeWithoutHeader,
- long prevBlockOffset, ByteBuffer b, boolean fillHeader, long offset,
- final int nextBlockOnDiskSize, int onDiskDataSizeWithHeader, HFileContext fileContext) {
- init(blockType, onDiskSizeWithoutHeader, uncompressedSizeWithoutHeader,
- prevBlockOffset, offset, onDiskDataSizeWithHeader, nextBlockOnDiskSize, fileContext);
- this.buf = new SingleByteBuff(b);
+ long prevBlockOffset, ByteBuff buf, boolean fillHeader, long offset,
+ int onDiskDataSizeWithHeader, HFileContext fileContext) {
+ this.blockType = blockType;
+ this.onDiskSizeWithoutHeader = onDiskSizeWithoutHeader;
+ this.uncompressedSizeWithoutHeader = uncompressedSizeWithoutHeader;
+ this.prevBlockOffset = prevBlockOffset;
+ this.buf = buf;
+ this.offset = offset;
+ this.onDiskDataSizeWithHeader = onDiskDataSizeWithHeader;
+ this.fileContext = fileContext;
if (fillHeader) {
overwriteHeader();
}
this.buf.rewind();
}
- /**
- * Creates a block from an existing buffer starting with a header. Rewinds
- * and takes ownership of the buffer. By definition of rewind, ignores the
- * buffer position, but if you slice the buffer beforehand, it will rewind
- * to that point.
- * @param buf Has header, content, and trailing checksums if present.
- */
- HFileBlock(ByteBuff buf, boolean usesHBaseChecksum, MemoryType memType, final long offset,
- final int nextBlockOnDiskSize, HFileContext fileContext) throws IOException {
- buf.rewind();
- final BlockType blockType = BlockType.read(buf);
- final int onDiskSizeWithoutHeader = buf.getInt();
- final int uncompressedSizeWithoutHeader = buf.getInt();
- final long prevBlockOffset = buf.getLong();
- byte checksumType = buf.get();
- int bytesPerChecksum = buf.getInt();
- int onDiskDataSizeWithHeader = buf.getInt();
- // This constructor is called when we deserialize a block from cache and when we read a block in
- // from the fs. fileCache is null when deserialized from cache so need to make up one.
- HFileContextBuilder fileContextBuilder = fileContext != null?
- new HFileContextBuilder(fileContext): new HFileContextBuilder();
- fileContextBuilder.withHBaseCheckSum(usesHBaseChecksum);
- if (usesHBaseChecksum) {
- // Use the checksum type and bytes per checksum from header, not from filecontext.
- fileContextBuilder.withChecksumType(ChecksumType.codeToType(checksumType));
- fileContextBuilder.withBytesPerCheckSum(bytesPerChecksum);
- } else {
- fileContextBuilder.withChecksumType(ChecksumType.NULL);
- fileContextBuilder.withBytesPerCheckSum(0);
- // Need to fix onDiskDataSizeWithHeader; there are not checksums after-block-data
- onDiskDataSizeWithHeader = onDiskSizeWithoutHeader + headerSize(usesHBaseChecksum);
- }
- fileContext = fileContextBuilder.build();
- assert usesHBaseChecksum == fileContext.isUseHBaseChecksum();
- init(blockType, onDiskSizeWithoutHeader, uncompressedSizeWithoutHeader,
- prevBlockOffset, offset, onDiskDataSizeWithHeader, nextBlockOnDiskSize, fileContext);
- this.memType = memType;
- this.offset = offset;
- this.buf = buf;
- this.buf.rewind();
+ HFileBlock(BlockType blockType, int onDiskSizeWithoutHeader, int uncompressedSizeWithoutHeader,
+ long prevBlockOffset, ByteBuffer buf, boolean fillHeader, long offset,
+ int onDiskDataSizeWithHeader, HFileContext fileContext) {
+ this(blockType, onDiskSizeWithoutHeader, uncompressedSizeWithoutHeader, prevBlockOffset,
+ new SingleByteBuff(buf), fillHeader, offset, onDiskDataSizeWithHeader, fileContext);
}
/**
- * Called from constructors.
+ * Copy constructor. Creates a shallow copy of {@code that}'s buffer.
*/
- private void init(BlockType blockType, int onDiskSizeWithoutHeader,
- int uncompressedSizeWithoutHeader, long prevBlockOffset,
- long offset, int onDiskDataSizeWithHeader, final int nextBlockOnDiskSize,
- HFileContext fileContext) {
- this.blockType = blockType;
- this.onDiskSizeWithoutHeader = onDiskSizeWithoutHeader;
- this.uncompressedSizeWithoutHeader = uncompressedSizeWithoutHeader;
- this.prevBlockOffset = prevBlockOffset;
- this.offset = offset;
- this.onDiskDataSizeWithHeader = onDiskDataSizeWithHeader;
- this.nextBlockOnDiskSize = nextBlockOnDiskSize;
- this.fileContext = fileContext;
+ HFileBlock(HFileBlock that) {
+ this.blockType = that.blockType;
+ this.onDiskSizeWithoutHeader = that.onDiskSizeWithoutHeader;
+ this.uncompressedSizeWithoutHeader = that.uncompressedSizeWithoutHeader;
+ this.prevBlockOffset = that.prevBlockOffset;
+ this.buf = that.buf.duplicate();
+ this.offset = that.offset;
+ this.onDiskDataSizeWithHeader = that.onDiskDataSizeWithHeader;
+ this.fileContext = that.fileContext;
+ this.nextBlockOnDiskSizeWithHeader = that.nextBlockOnDiskSizeWithHeader;
+ }
+
+ HFileBlock(ByteBuffer b, boolean usesHBaseChecksum) throws IOException {
+ this(new SingleByteBuff(b), usesHBaseChecksum);
}
/**
- * Parse total ondisk size including header and checksum. Its second field in header after
- * the magic bytes.
- * @param headerBuf Header ByteBuffer. Presumed exact size of header.
- * @return Size of the block with header included.
+ * Creates a block from an existing buffer starting with a header. Rewinds
+ * and takes ownership of the buffer. By definition of rewind, ignores the
+ * buffer position, but if you slice the buffer beforehand, it will rewind
+ * to that point.
*/
- private static int getOnDiskSizeWithHeader(final ByteBuffer headerBuf) {
- // Set hbase checksum to true always calling headerSize.
- return headerBuf.getInt(BlockType.MAGIC_LENGTH) + headerSize(true);
+ HFileBlock(ByteBuff b, boolean usesHBaseChecksum) throws IOException {
+ this(b, usesHBaseChecksum, MemoryType.EXCLUSIVE);
}
/**
- * @return the on-disk size of the next block (including the header size and any checksums if
- * present) read by peeking into the next block's header; use as a hint when doing
- * a read of the next block when scanning or running over a file.
+ * Creates a block from an existing buffer starting with a header. Rewinds
+ * and takes ownership of the buffer. By definition of rewind, ignores the
+ * buffer position, but if you slice the buffer beforehand, it will rewind
+ * to that point.
*/
- public int getNextBlockOnDiskSize() {
- return nextBlockOnDiskSize;
+ HFileBlock(ByteBuff b, boolean usesHBaseChecksum, MemoryType memType) throws IOException {
+ b.rewind();
+ blockType = BlockType.read(b);
+ onDiskSizeWithoutHeader = b.getInt();
+ uncompressedSizeWithoutHeader = b.getInt();
+ prevBlockOffset = b.getLong();
+ HFileContextBuilder contextBuilder = new HFileContextBuilder();
+ contextBuilder.withHBaseCheckSum(usesHBaseChecksum);
+ if (usesHBaseChecksum) {
+ contextBuilder.withChecksumType(ChecksumType.codeToType(b.get()));
+ contextBuilder.withBytesPerCheckSum(b.getInt());
+ this.onDiskDataSizeWithHeader = b.getInt();
+ } else {
+ contextBuilder.withChecksumType(ChecksumType.NULL);
+ contextBuilder.withBytesPerCheckSum(0);
+ this.onDiskDataSizeWithHeader =
+ onDiskSizeWithoutHeader + HConstants.HFILEBLOCK_HEADER_SIZE_NO_CHECKSUM;
+ }
+ this.fileContext = contextBuilder.build();
+ this.memType = memType;
+ buf = b;
+ buf.rewind();
}
public BlockType getBlockType() {
@@ -468,26 +414,49 @@ public class HFileBlock implements Cacheable {
* @return the buffer with header skipped and checksum omitted.
*/
public ByteBuff getBufferWithoutHeader() {
- ByteBuff dup = getBufferReadOnly();
- // Now set it up so Buffer spans content only -- no header or no checksums.
- return dup.position(headerSize()).limit(buf.limit() - totalChecksumBytes()).slice();
+ ByteBuff dup = this.buf.duplicate();
+ dup.position(headerSize());
+ dup.limit(buf.limit() - totalChecksumBytes());
+ return dup.slice();
}
/**
- * Returns a read-only duplicate of the buffer this block stores internally ready to be read.
- * Clients must not modify the buffer object though they may set position and limit on the
- * returned buffer since we pass back a duplicate. This method has to be public because it is used
+ * Returns the buffer this block stores internally. The clients must not
+ * modify the buffer object. This method has to be public because it is used
* in {@link CompoundBloomFilter} to avoid object creation on every Bloom
- * filter lookup, but has to be used with caution. Buffer holds header, block content,
- * and any follow-on checksums if present.
+ * filter lookup, but has to be used with caution. Checksum data is not
+ * included in the returned buffer but header data is.
*
* @return the buffer of this block for read-only operations
*/
- public ByteBuff getBufferReadOnly() {
- // TODO: ByteBuf does not support asReadOnlyBuffer(). Fix.
+ ByteBuff getBufferReadOnly() {
+ ByteBuff dup = this.buf.duplicate();
+ dup.limit(buf.limit() - totalChecksumBytes());
+ return dup.slice();
+ }
+
+ /**
+ * Returns the buffer of this block, including header data. The clients must
+ * not modify the buffer object. This method has to be public because it is
+ * used in {@link org.apache.hadoop.hbase.io.hfile.bucket.BucketCache} to avoid buffer copy.
+ *
+ * @return the buffer with header and checksum included for read-only operations
+ */
+ public ByteBuff getBufferReadOnlyWithHeader() {
ByteBuff dup = this.buf.duplicate();
- assert dup.position() == 0;
- return dup;
+ return dup.slice();
+ }
+
+ /**
+ * Returns a byte buffer of this block, including header data and checksum, positioned at
+ * the beginning of header. The underlying data array is not copied.
+ *
+ * @return the byte buffer with header and checksum included
+ */
+ ByteBuff getBufferWithHeader() {
+ ByteBuff dupBuf = buf.duplicate();
+ dupBuf.rewind();
+ return dupBuf;
}
private void sanityCheckAssertion(long valueFromBuf, long valueFromField,
@@ -512,38 +481,39 @@ public class HFileBlock implements Cacheable {
* valid header consistent with the fields. Assumes a packed block structure.
* This function is primary for testing and debugging, and is not
* thread-safe, because it alters the internal buffer pointer.
- * Used by tests only.
*/
- @VisibleForTesting
void sanityCheck() throws IOException {
- // Duplicate so no side-effects
- ByteBuff dup = this.buf.duplicate().rewind();
- sanityCheckAssertion(BlockType.read(dup), blockType);
+ buf.rewind();
+
+ sanityCheckAssertion(BlockType.read(buf), blockType);
- sanityCheckAssertion(dup.getInt(), onDiskSizeWithoutHeader, "onDiskSizeWithoutHeader");
+ sanityCheckAssertion(buf.getInt(), onDiskSizeWithoutHeader,
+ "onDiskSizeWithoutHeader");
- sanityCheckAssertion(dup.getInt(), uncompressedSizeWithoutHeader,
+ sanityCheckAssertion(buf.getInt(), uncompressedSizeWithoutHeader,
"uncompressedSizeWithoutHeader");
- sanityCheckAssertion(dup.getLong(), prevBlockOffset, "prevBlockOffset");
+ sanityCheckAssertion(buf.getLong(), prevBlockOffset, "prevBlocKOffset");
if (this.fileContext.isUseHBaseChecksum()) {
- sanityCheckAssertion(dup.get(), this.fileContext.getChecksumType().getCode(), "checksumType");
- sanityCheckAssertion(dup.getInt(), this.fileContext.getBytesPerChecksum(),
+ sanityCheckAssertion(buf.get(), this.fileContext.getChecksumType().getCode(), "checksumType");
+ sanityCheckAssertion(buf.getInt(), this.fileContext.getBytesPerChecksum(),
"bytesPerChecksum");
- sanityCheckAssertion(dup.getInt(), onDiskDataSizeWithHeader, "onDiskDataSizeWithHeader");
+ sanityCheckAssertion(buf.getInt(), onDiskDataSizeWithHeader, "onDiskDataSizeWithHeader");
}
int cksumBytes = totalChecksumBytes();
int expectedBufLimit = onDiskDataSizeWithHeader + cksumBytes;
- if (dup.limit() != expectedBufLimit) {
- throw new AssertionError("Expected limit " + expectedBufLimit + ", got " + dup.limit());
+ if (buf.limit() != expectedBufLimit) {
+ throw new AssertionError("Expected buffer limit " + expectedBufLimit
+ + ", got " + buf.limit());
}
// We might optionally allocate HFILEBLOCK_HEADER_SIZE more bytes to read the next
// block's header, so there are two sensible values for buffer capacity.
int hdrSize = headerSize();
- if (dup.capacity() != expectedBufLimit && dup.capacity() != expectedBufLimit + hdrSize) {
- throw new AssertionError("Invalid buffer capacity: " + dup.capacity() +
+ if (buf.capacity() != expectedBufLimit &&
+ buf.capacity() != expectedBufLimit + hdrSize) {
+ throw new AssertionError("Invalid buffer capacity: " + buf.capacity() +
", expected " + expectedBufLimit + " or " + (expectedBufLimit + hdrSize));
}
}
@@ -590,6 +560,30 @@ public class HFileBlock implements Cacheable {
}
/**
+ * Called after reading a block with provided onDiskSizeWithHeader.
+ */
+ private void validateOnDiskSizeWithoutHeader(int expectedOnDiskSizeWithoutHeader)
+ throws IOException {
+ if (onDiskSizeWithoutHeader != expectedOnDiskSizeWithoutHeader) {
+ String dataBegin = null;
+ if (buf.hasArray()) {
+ dataBegin = Bytes.toStringBinary(buf.array(), buf.arrayOffset(), Math.min(32, buf.limit()));
+ } else {
+ ByteBuff bufDup = getBufferReadOnly();
+ byte[] dataBeginBytes = new byte[Math.min(32, bufDup.limit() - bufDup.position())];
+ bufDup.get(dataBeginBytes);
+ dataBegin = Bytes.toStringBinary(dataBeginBytes);
+ }
+ String blockInfoMsg =
+ "Block offset: " + offset + ", data starts with: " + dataBegin;
+ throw new IOException("On-disk size without header provided is "
+ + expectedOnDiskSizeWithoutHeader + ", but block "
+ + "header contains " + onDiskSizeWithoutHeader + ". " +
+ blockInfoMsg);
+ }
+ }
+
+ /**
* Retrieves the decompressed/decrypted view of this block. An encoded block remains in its
* encoded structure. Internal structures are shared between instances where applicable.
*/
@@ -613,10 +607,33 @@ public class HFileBlock implements Cacheable {
ctx.prepareDecoding(unpacked.getOnDiskSizeWithoutHeader(),
unpacked.getUncompressedSizeWithoutHeader(), unpacked.getBufferWithoutHeader(),
dup);
+
+ // Preserve the next block's header bytes in the new block if we have them.
+ if (unpacked.hasNextBlockHeader()) {
+ // Both the buffers are limited till checksum bytes and avoid the next block's header.
+ // Below call to copyFromBufferToBuffer() will try positional read/write from/to buffers when
+ // any of the buffer is DBB. So we change the limit on a dup buffer. No copying just create
+ // new BB objects
+ ByteBuff inDup = this.buf.duplicate();
+ inDup.limit(inDup.limit() + headerSize());
+ ByteBuff outDup = unpacked.buf.duplicate();
+ outDup.limit(outDup.limit() + unpacked.headerSize());
+ outDup.put(
+ unpacked.headerSize() + unpacked.uncompressedSizeWithoutHeader
+ + unpacked.totalChecksumBytes(), inDup, this.onDiskDataSizeWithHeader,
+ unpacked.headerSize());
+ }
return unpacked;
}
/**
+ * Return true when this buffer includes next block's header.
+ */
+ private boolean hasNextBlockHeader() {
+ return nextBlockOnDiskSizeWithHeader > 0;
+ }
+
+ /**
* Always allocates a new buffer of the correct size. Copies header bytes
* from the existing buffer. Does not change header fields.
* Reserve room to keep checksum bytes too.
@@ -624,7 +641,8 @@ public class HFileBlock implements Cacheable {
private void allocateBuffer() {
int cksumBytes = totalChecksumBytes();
int headerSize = headerSize();
- int capacityNeeded = headerSize + uncompressedSizeWithoutHeader + cksumBytes;
+ int capacityNeeded = headerSize + uncompressedSizeWithoutHeader +
+ cksumBytes + (hasNextBlockHeader() ? headerSize : 0);
// TODO we need consider allocating offheap here?
ByteBuffer newBuf = ByteBuffer.allocate(capacityNeeded);
@@ -652,8 +670,9 @@ public class HFileBlock implements Cacheable {
}
/** An additional sanity-check in case no compression or encryption is being used. */
- public void sanityCheckUncompressedSize() throws IOException {
- if (onDiskSizeWithoutHeader != uncompressedSizeWithoutHeader + totalChecksumBytes()) {
+ public void assumeUncompressed() throws IOException {
+ if (onDiskSizeWithoutHeader != uncompressedSizeWithoutHeader +
+ totalChecksumBytes()) {
throw new IOException("Using no compression but "
+ "onDiskSizeWithoutHeader=" + onDiskSizeWithoutHeader + ", "
+ "uncompressedSizeWithoutHeader=" + uncompressedSizeWithoutHeader
@@ -661,14 +680,11 @@ public class HFileBlock implements Cacheable {
}
}
- /**
- * Cannot be {@link #UNSET}. Must be a legitimate value. Used re-making the {@link CacheKey} when
- * block is returned to the cache.
- * @return the offset of this block in the file it was read from
- */
+ /** @return the offset of this block in the file it was read from */
long getOffset() {
if (offset < 0) {
- throw new IllegalStateException("HFile block offset not initialized properly");
+ throw new IllegalStateException(
+ "HFile block offset not initialized properly");
}
return offset;
}
@@ -728,6 +744,7 @@ public class HFileBlock implements Cacheable {
// We could not read the "extra data", but that is OK.
break;
}
+
if (ret < 0) {
throw new IOException("Premature EOF from inputStream (read "
+ "returned " + ret + ", was trying to read " + necessaryLen
@@ -782,6 +799,14 @@ public class HFileBlock implements Cacheable {
}
/**
+ * @return the on-disk size of the next block (including the header size)
+ * that was read by peeking into the next block's header
+ */
+ public int getNextBlockOnDiskSizeWithHeader() {
+ return nextBlockOnDiskSizeWithHeader;
+ }
+
+ /**
* Unified version 2 {@link HFile} block writer. The intended usage pattern
* is as follows:
* <ol>
@@ -813,8 +838,8 @@ public class HFileBlock implements Cacheable {
private HFileBlockDefaultEncodingContext defaultBlockEncodingCtx;
/**
- * The stream we use to accumulate data into a block in an uncompressed format.
- * We reset this stream at the end of each block and reuse it. The
+ * The stream we use to accumulate data in uncompressed format for each
+ * block. We reset this stream at the end of each block and reuse it. The
* header is written as the first {@link HConstants#HFILEBLOCK_HEADER_SIZE} bytes into this
* stream.
*/
@@ -842,7 +867,7 @@ public class HFileBlock implements Cacheable {
* if compression is turned on. It also includes the checksum data that
* immediately follows the block data. (header + data + checksums)
*/
- private byte[] onDiskBlockBytesWithHeader;
+ private byte[] onDiskBytesWithHeader;
/**
* The size of the checksum data on disk. It is used only if data is
@@ -859,7 +884,7 @@ public class HFileBlock implements Cacheable {
* {@link org.apache.hadoop.hbase.HConstants#HFILEBLOCK_HEADER_SIZE}.
* Does not store checksums.
*/
- private byte[] uncompressedBlockBytesWithHeader;
+ private byte[] uncompressedBytesWithHeader;
/**
* Current block's start offset in the {@link HFile}. Set in
@@ -967,19 +992,18 @@ public class HFileBlock implements Cacheable {
Preconditions.checkState(state != State.INIT,
"Unexpected state: " + state);
- if (state == State.BLOCK_READY) {
+ if (state == State.BLOCK_READY)
return;
- }
// This will set state to BLOCK_READY.
finishBlock();
}
/**
- * Finish up writing of the block.
- * Flushes the compressing stream (if using compression), fills out the header,
- * does any compression/encryption of bytes to flush out to disk, and manages
- * the cache on write content, if applicable. Sets block write state to "block ready".
+ * An internal method that flushes the compressing stream (if using
+ * compression), serializes the header, and takes care of the separate
+ * uncompressed stream for caching on write, if applicable. Sets block
+ * write state to "block ready".
*/
private void finishBlock() throws IOException {
if (blockType == BlockType.DATA) {
@@ -988,40 +1012,41 @@ public class HFileBlock implements Cacheable {
blockType = dataBlockEncodingCtx.getBlockType();
}
userDataStream.flush();
- // This does an array copy, so it is safe to cache this byte array when cache-on-write.
+ // This does an array copy, so it is safe to cache this byte array.
// Header is still the empty, 'dummy' header that is yet to be filled out.
- uncompressedBlockBytesWithHeader = baosInMemory.toByteArray();
+ uncompressedBytesWithHeader = baosInMemory.toByteArray();
prevOffset = prevOffsetByType[blockType.getId()];
- // We need to set state before we can package the block up for cache-on-write. In a way, the
- // block is ready, but not yet encoded or compressed.
+ // We need to set state before we can package the block up for
+ // cache-on-write. In a way, the block is ready, but not yet encoded or
+ // compressed.
state = State.BLOCK_READY;
if (blockType == BlockType.DATA || blockType == BlockType.ENCODED_DATA) {
- onDiskBlockBytesWithHeader = dataBlockEncodingCtx.
- compressAndEncrypt(uncompressedBlockBytesWithHeader);
+ onDiskBytesWithHeader = dataBlockEncodingCtx
+ .compressAndEncrypt(uncompressedBytesWithHeader);
} else {
- onDiskBlockBytesWithHeader = defaultBlockEncodingCtx.
- compressAndEncrypt(uncompressedBlockBytesWithHeader);
+ onDiskBytesWithHeader = this.defaultBlockEncodingCtx.
+ compressAndEncrypt(uncompressedBytesWithHeader);
}
// Calculate how many bytes we need for checksum on the tail of the block.
int numBytes = (int) ChecksumUtil.numBytes(
- onDiskBlockBytesWithHeader.length,
+ onDiskBytesWithHeader.length,
fileContext.getBytesPerChecksum());
// Put the header for the on disk bytes; header currently is unfilled-out
- putHeader(onDiskBlockBytesWithHeader, 0,
- onDiskBlockBytesWithHeader.length + numBytes,
- uncompressedBlockBytesWithHeader.length, onDiskBlockBytesWithHeader.length);
+ putHeader(onDiskBytesWithHeader, 0,
+ onDiskBytesWithHeader.length + numBytes,
+ uncompressedBytesWithHeader.length, onDiskBytesWithHeader.length);
// Set the header for the uncompressed bytes (for cache-on-write) -- IFF different from
- // onDiskBlockBytesWithHeader array.
- if (onDiskBlockBytesWithHeader != uncompressedBlockBytesWithHeader) {
- putHeader(uncompressedBlockBytesWithHeader, 0,
- onDiskBlockBytesWithHeader.length + numBytes,
- uncompressedBlockBytesWithHeader.length, onDiskBlockBytesWithHeader.length);
+ // onDiskBytesWithHeader array.
+ if (onDiskBytesWithHeader != uncompressedBytesWithHeader) {
+ putHeader(uncompressedBytesWithHeader, 0,
+ onDiskBytesWithHeader.length + numBytes,
+ uncompressedBytesWithHeader.length, onDiskBytesWithHeader.length);
}
onDiskChecksum = new byte[numBytes];
ChecksumUtil.generateChecksums(
- onDiskBlockBytesWithHeader, 0, onDiskBlockBytesWithHeader.length,
+ onDiskBytesWithHeader, 0, onDiskBytesWithHeader.length,
onDiskChecksum, 0, fileContext.getChecksumType(), fileContext.getBytesPerChecksum());
}
@@ -1076,7 +1101,7 @@ public class HFileBlock implements Cacheable {
protected void finishBlockAndWriteHeaderAndData(DataOutputStream out)
throws IOException {
ensureBlockReady();
- out.write(onDiskBlockBytesWithHeader);
+ out.write(onDiskBytesWithHeader);
out.write(onDiskChecksum);
}
@@ -1095,12 +1120,12 @@ public class HFileBlock implements Cacheable {
// This is not very optimal, because we are doing an extra copy.
// But this method is used only by unit tests.
byte[] output =
- new byte[onDiskBlockBytesWithHeader.length
+ new byte[onDiskBytesWithHeader.length
+ onDiskChecksum.length];
- System.arraycopy(onDiskBlockBytesWithHeader, 0, output, 0,
- onDiskBlockBytesWithHeader.length);
+ System.arraycopy(onDiskBytesWithHeader, 0, output, 0,
+ onDiskBytesWithHeader.length);
System.arraycopy(onDiskChecksum, 0, output,
- onDiskBlockBytesWithHeader.length, onDiskChecksum.length);
+ onDiskBytesWithHeader.length, onDiskChecksum.length);
return output;
}
@@ -1128,7 +1153,7 @@ public class HFileBlock implements Cacheable {
*/
int getOnDiskSizeWithoutHeader() {
expectState(State.BLOCK_READY);
- return onDiskBlockBytesWithHeader.length +
+ return onDiskBytesWithHeader.length +
onDiskChecksum.length - HConstants.HFILEBLOCK_HEADER_SIZE;
}
@@ -1141,7 +1166,7 @@ public class HFileBlock implements Cacheable {
*/
int getOnDiskSizeWithHeader() {
expectState(State.BLOCK_READY);
- return onDiskBlockBytesWithHeader.length + onDiskChecksum.length;
+ return onDiskBytesWithHeader.length + onDiskChecksum.length;
}
/**
@@ -1149,7 +1174,7 @@ public class HFileBlock implements Cacheable {
*/
int getUncompressedSizeWithoutHeader() {
expectState(State.BLOCK_READY);
- return uncompressedBlockBytesWithHeader.length - HConstants.HFILEBLOCK_HEADER_SIZE;
+ return uncompressedBytesWithHeader.length - HConstants.HFILEBLOCK_HEADER_SIZE;
}
/**
@@ -1157,7 +1182,7 @@ public class HFileBlock implements Cacheable {
*/
int getUncompressedSizeWithHeader() {
expectState(State.BLOCK_READY);
- return uncompressedBlockBytesWithHeader.length;
+ return uncompressedBytesWithHeader.length;
}
/** @return true if a block is being written */
@@ -1187,7 +1212,7 @@ public class HFileBlock implements Cacheable {
*/
ByteBuffer getUncompressedBufferWithHeader() {
expectState(State.BLOCK_READY);
- return ByteBuffer.wrap(uncompressedBlockBytesWithHeader);
+ return ByteBuffer.wrap(uncompressedBytesWithHeader);
}
/**
@@ -1200,7 +1225,7 @@ public class HFileBlock implements Cacheable {
*/
ByteBuffer getOnDiskBufferWithHeader() {
expectState(State.BLOCK_READY);
- return ByteBuffer.wrap(onDiskBlockBytesWithHeader);
+ return ByteBuffer.wrap(onDiskBytesWithHeader);
}
private void expectState(State expectedState) {
@@ -1232,10 +1257,6 @@ public class HFileBlock implements Cacheable {
* block does not have checksum data even though the header minor
* version is MINOR_VERSION_WITH_CHECKSUM. This is indicated by setting a
* 0 value in bytesPerChecksum.
- *
- * <p>TODO: Should there be an option where a cache can ask that hbase preserve block
- * checksums for checking after a block comes out of the cache? Otehrwise, cache is responsible
- * for blocks being wholesome (ECC memory or if file-backed, it does checksumming).
*/
HFileBlock getBlockForCaching(CacheConfig cacheConf) {
HFileContext newContext = new HFileContextBuilder()
@@ -1249,13 +1270,13 @@ public class HFileBlock implements Cacheable {
.withIncludesMvcc(fileContext.isIncludesMvcc())
.withIncludesTags(fileContext.isIncludesTags())
.build();
- return new HFileBlock(blockType, getOnDiskSizeWithoutHeader(),
+ return new HFileBlock(blockType, getOnDiskSizeWithoutHeader(),
getUncompressedSizeWithoutHeader(), prevOffset,
- cacheConf.shouldCacheCompressed(blockType.getCategory())?
+ cacheConf.shouldCacheCompressed(blockType.getCategory()) ?
getOnDiskBufferWithHeader() :
getUncompressedBufferWithHeader(),
- FILL_HEADER, startOffset, UNSET,
- onDiskBlockBytesWithHeader.length + onDiskChecksum.length, newContext);
+ FILL_HEADER, startOffset,
+ onDiskBytesWithHeader.length + onDiskChecksum.length, newContext);
}
}
@@ -1301,9 +1322,12 @@ public class HFileBlock implements Cacheable {
* @param offset
* @param onDiskSize the on-disk size of the entire block, including all
* applicable headers, or -1 if unknown
+ * @param uncompressedSize the uncompressed size of the compressed part of
+ * the block, or -1 if unknown
* @return the newly read block
*/
- HFileBlock readBlockData(long offset, long onDiskSize, boolean pread) throws IOException;
+ HFileBlock readBlockData(long offset, long onDiskSize,
+ int uncompressedSize, boolean pread) throws IOException;
/**
* Creates a block iterator over the given portion of the {@link HFile}.
@@ -1356,11 +1380,6 @@ public class HFileBlock implements Cacheable {
/** Default context used when BlockType != {@link BlockType#ENCODED_DATA}. */
private final HFileBlockDefaultDecodingContext defaultDecodingCtx;
- /**
- * When we read a block, we overread and pull in the next blocks header too. We will save it
- * here. If moving serially through the file, we will trip over this caching of the next blocks
- * header so we won't have to do explicit seek to find next blocks lengths, etc.
- */
private ThreadLocal<PrefetchedHeader> prefetchedHeaderForThread =
new ThreadLocal<PrefetchedHeader>() {
@Override
@@ -1424,7 +1443,7 @@ public class HFileBlock implements Cacheable {
public HFileBlock nextBlock() throws IOException {
if (offset >= endOffset)
return null;
- HFileBlock b = readBlockData(offset, -1, false);
+ HFileBlock b = readBlockData(offset, -1, -1, false);
offset += b.getOnDiskSizeWithHeader();
return b.unpack(fileContext, owner);
}
@@ -1444,7 +1463,7 @@ public class HFileBlock implements Cacheable {
/**
* Does a positional read or a seek and read into the given buffer. Returns
- * the on-disk size of the next block, or -1 if it could not be read/determined; e.g. EOF.
+ * the on-disk size of the next block, or -1 if it could not be determined.
*
* @param dest destination buffer
* @param destOffset offset into the destination buffer at where to put the bytes we read
@@ -1454,8 +1473,7 @@ public class HFileBlock implements Cacheable {
* @param pread whether we should do a positional read
* @param istream The input source of data
* @return the on-disk size of the next block with header size included, or
- * -1 if it could not be determined; if not -1, the <code>dest</code> INCLUDES the
- * next header
+ * -1 if it could not be determined
* @throws IOException
*/
protected int readAtOffset(FSDataInputStream istream, byte [] dest, int destOffset, int size,
@@ -1487,16 +1505,16 @@ public class HFileBlock implements Cacheable {
}
// Try to read the next block header.
- if (!readWithExtra(istream, dest, destOffset, size, hdrSize)) {
+ if (!readWithExtra(istream, dest, destOffset, size, hdrSize))
return -1;
- }
} finally {
streamLock.unlock();
}
} else {
// Positional read. Better for random reads; or when the streamLock is already locked.
int extraSize = peekIntoNextBlock ? hdrSize : 0;
- if (!positionalReadWithExtra(istream, fileOffset, dest, destOffset, size, extraSize)) {
+ if (!positionalReadWithExtra(istream, fileOffset, dest, destOffset,
+ size, extraSize)) {
return -1;
}
}
@@ -1512,12 +1530,16 @@ public class HFileBlock implements Cacheable {
* @param offset the offset in the stream to read at
* @param onDiskSizeWithHeaderL the on-disk size of the block, including
* the header, or -1 if unknown
+ * @param uncompressedSize the uncompressed size of the the block. Always
+ * expected to be -1. This parameter is only used in version 1.
* @param pread whether to use a positional read
*/
@Override
- public HFileBlock readBlockData(long offset, long onDiskSizeWithHeaderL, boolean pread)
+ public HFileBlock readBlockData(long offset, long onDiskSizeWithHeaderL,
+ int uncompressedSize, boolean pread)
throws IOException {
- // Get a copy of the current state of whether to validate
+
+ // get a copy of the current state of whether to validate
// hbase checksums or not for this read call. This is not
// thread-safe but the one constaint is that if we decide
// to skip hbase checksum verification then we are
@@ -1526,7 +1548,8 @@ public class HFileBlock implements Cacheable {
FSDataInputStream is = streamWrapper.getStream(doVerificationThruHBaseChecksum);
HFileBlock blk = readBlockDataInternal(is, offset,
- onDiskSizeWithHeaderL, pread,
+ onDiskSizeWithHeaderL,
+ uncompressedSize, pread,
doVerificationThruHBaseChecksum);
if (blk == null) {
HFile.LOG.warn("HBase checksum verification failed for file " +
@@ -1553,7 +1576,8 @@ public class HFileBlock implements Cacheable {
// a few more than precisely this number.
is = this.streamWrapper.fallbackToFsChecksum(CHECKSUM_VERIFICATION_NUM_IO_THRESHOLD);
doVerificationThruHBaseChecksum = false;
- blk = readBlockDataInternal(is, offset, onDiskSizeWithHeaderL, pread,
+ blk = readBlockDataInternal(is, offset, onDiskSizeWithHeaderL,
+ uncompressedSize, pread,
doVerificationThruHBaseChecksum);
if (blk != null) {
HFile.LOG.warn("HDFS checksum verification suceeded for file " +
@@ -1581,139 +1605,175 @@ public class HFileBlock implements Cacheable {
}
/**
- * @return Check <code>onDiskSizeWithHeaderL</code> size is healthy and then return it as an int
- * @throws IOException
- */
- private static int checkAndGetSizeAsInt(final long onDiskSizeWithHeaderL, final int hdrSize)
- throws IOException {
- if ((onDiskSizeWithHeaderL < hdrSize && onDiskSizeWithHeaderL != -1)
- || onDiskSizeWithHeaderL >= Integer.MAX_VALUE) {
- throw new IOException("Invalid onDisksize=" + onDiskSizeWithHeaderL
- + ": expected to be at least " + hdrSize
- + " and at most " + Integer.MAX_VALUE + ", or -1");
- }
- return (int)onDiskSizeWithHeaderL;
- }
-
- /**
- * Check threadlocal cache for this block's header; we usually read it on the tail of reading
- * the previous block to save a seek. Otherwise, we have to do a seek to read the header before
- * we can pull in the block.
- * @return The cached block header or null if not found.
- * @see #cacheNextBlockHeader(long, byte[], int, int)
- */
- private ByteBuffer getCachedHeader(final long offset) {
- PrefetchedHeader prefetchedHeader = prefetchedHeaderForThread.get();
- // PrefetchedHeader prefetchedHeader = prefetchedHeaderForThread.get();
- return prefetchedHeader != null && prefetchedHeader.offset == offset?
- prefetchedHeader.buf: null;
- }
-
- /**
- * Save away the next blocks header in thread local.
- * @see #getCachedHeader(long)
- */
- private void cacheNextBlockHeader(final long nextBlockOffset,
- final byte [] header, final int headerOffset, final int headerLength) {
- PrefetchedHeader prefetchedHeader = prefetchedHeaderForThread.get();
- prefetchedHeader.offset = nextBlockOffset;
- System.arraycopy(header, headerOffset, prefetchedHeader.header, 0, headerLength);
- }
-
- /**
- * Verify the passed in onDiskSizeWithHeader aligns with what is in the header else something
- * is not right.
- * @throws IOException
- */
- private void verifyOnDiskSizeMatchesHeader(final int passedIn, final ByteBuffer headerBuf,
- final long offset)
- throws IOException {
- // Assert size provided aligns with what is in the header
- int fromHeader = getOnDiskSizeWithHeader(headerBuf);
- if (passedIn != fromHeader) {
- throw new IOException("Passed in onDiskSizeWithHeader=" + passedIn + " != " + fromHeader +
- ", offset=" + offset + ", fileContext=" + this.fileContext);
- }
- }
-
- /**
* Reads a version 2 block.
*
* @param offset the offset in the stream to read at
* @param onDiskSizeWithHeaderL the on-disk size of the block, including
- * the header and checksums if present or -1 if unknown
+ * the header, or -1 if unknown
+ * @param uncompressedSize the uncompressed size of the the block. Always
+ * expected to be -1. This parameter is only used in version 1.
* @param pread whether to use a positional read
* @param verifyChecksum Whether to use HBase checksums.
* If HBase checksum is switched off, then use HDFS checksum.
* @return the HFileBlock or null if there is a HBase checksum mismatch
*/
private HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
- long onDiskSizeWithHeaderL, boolean pread, boolean verifyChecksum)
+ long onDiskSizeWithHeaderL, int uncompressedSize, boolean pread,
+ boolean verifyChecksum)
throws IOException {
if (offset < 0) {
throw new IOException("Invalid offset=" + offset + " trying to read "
- + "block (onDiskSize=" + onDiskSizeWithHeaderL + ")");
+ + "block (onDiskSize=" + onDiskSizeWithHeaderL
+ + ", uncompressedSize=" + uncompressedSize + ")");
}
- int onDiskSizeWithHeader = checkAndGetSizeAsInt(onDiskSizeWithHeaderL, hdrSize);
- ByteBuffer headerBuf = getCachedHeader(offset);
- if (LOG.isTraceEnabled()) {
- LOG.trace("Reading " + this.fileContext.getHFileName() + " at offset=" + offset +
- ", pread=" + pread + ", verifyChecksum=" + verifyChecksum + ", cachedHeader=" +
- headerBuf + ", onDiskSizeWithHeader=" + onDiskSizeWithHeader);
+
+ if (uncompressedSize != -1) {
+ throw new IOException("Version 2 block reader API does not need " +
+ "the uncompressed size parameter");
+ }
+
+ if ((onDiskSizeWithHeaderL < hdrSize && onDiskSizeWithHeaderL != -1)
+ || onDiskSizeWithHeaderL >= Integer.MAX_VALUE) {
+ throw new IOException("Invalid onDisksize=" + onDiskSizeWithHeaderL
+ + ": expected to be at least " + hdrSize
+ + " and at most " + Integer.MAX_VALUE + ", or -1 (offset="
+ + offset + ", uncompressedSize=" + uncompressedSize + ")");
+ }
+
+ int onDiskSizeWithHeader = (int) onDiskSizeWithHeaderL;
+
+ // See if we can avoid reading the header. This is desirable, because
+ // we will not incur a backward seek operation if we have already
+ // read this block's header as part of the previous read's look-ahead.
+ // And we also want to skip reading the header again if it has already
+ // been read.
+ // TODO: How often does this optimization fire? Has to be same thread so the thread local
+ // is pertinent and we have to be reading next block as in a big scan.
+ ByteBuffer headerBuf = null;
+ PrefetchedHeader prefetchedHeader = prefetchedHeaderForThread.get();
+ boolean preReadHeader = false;
+ if (prefetchedHeader != null && prefetchedHeader.offset == offset) {
+ headerBuf = prefetchedHeader.buf;
+ preReadHeader = true;
}
- if (onDiskSizeWithHeader <= 0) {
- // We were not passed the block size. Need to get it from the header. If header was not in
- // cache, need to seek to pull it in. This latter might happen when we are doing the first
- // read in a series of reads or a random read, and we don't have access to the block index.
- // This is costly and should happen very rarely.
+ // Allocate enough space to fit the next block's header too.
+ int nextBlockOnDiskSize = 0;
+ byte[] onDiskBlock = null;
+
+ HFileBlock b = null;
+ boolean fastPath = false;
+ boolean readHdrOnly = false;
+ if (onDiskSizeWithHeader > 0) {
+ fastPath = true;
+ // We know the total on-disk size. Read the entire block into memory,
+ // then parse the header. This code path is used when
+ // doing a random read operation relying on the block index, as well as
+ // when the client knows the on-disk size from peeking into the next
+ // block's header (e.g. this block's header) when reading the previous
+ // block. This is the faster and more preferable case.
+
+ // Size that we have to skip in case we have already read the header.
+ int preReadHeaderSize = headerBuf == null ? 0 : hdrSize;
+ onDiskBlock = new byte[onDiskSizeWithHeader + hdrSize]; // room for this block plus the
+ // next block's header
+ nextBlockOnDiskSize = readAtOffset(is, onDiskBlock,
+ preReadHeaderSize, onDiskSizeWithHeader - preReadHeaderSize,
+ true, offset + preReadHeaderSize, pread);
+ if (headerBuf != null) {
+ // the header has been read when reading the previous block, copy
+ // to this block's header
+ // headerBuf is HBB
+ assert headerBuf.hasArray();
+ System.arraycopy(headerBuf.array(),
+ headerBuf.arrayOffset(), onDiskBlock, 0, hdrSize);
+ } else {
+ headerBuf = ByteBuffer.wrap(onDiskBlock, 0, hdrSize);
+ }
+ // We know the total on-disk size but not the uncompressed size. Parse the header.
+ try {
+ // TODO: FIX!!! Expensive parse just to get a length
+ b = new HFileBlock(headerBuf, fileContext.isUseHBaseChecksum());
+ } catch (IOException ex) {
+ // Seen in load testing. Provide comprehensive debug info.
+ throw new IOException("Failed to read compressed block at "
+ + offset
+ + ", onDiskSizeWithoutHeader="
+ + onDiskSizeWithHeader
+ + ", preReadHeaderSize="
+ + hdrSize
+ + ", header.length="
+ + prefetchedHeader.header.length
+ + ", header bytes: "
+ + Bytes.toStringBinary(prefetchedHeader.header, 0,
+ hdrSize), ex);
+ }
+ // if the caller specifies a onDiskSizeWithHeader, validate it.
+ int onDiskSizeWithoutHeader = onDiskSizeWithHeader - hdrSize;
+ assert onDiskSizeWithoutHeader >= 0;
+ b.validateOnDiskSizeWithoutHeader(onDiskSizeWithoutHeader);
+ } else {
+ // Check headerBuf to see if we have read this block's header as part of
+ // reading the previous block. This is an optimization of peeking into
+ // the next block's header (e.g.this block's header) when reading the
+ // previous block. This is the faster and more preferable case. If the
+ // header is already there, don't read the header again.
+
+ // Unfortunately, we still have to do a separate read operation to
+ // read the header.
if (headerBuf == null) {
+ readHdrOnly = true;
+ // From the header, determine the on-disk size of the given hfile
+ // block, and read the remaining data, thereby incurring two read
+ // operations. This might happen when we are doing the first read
+ // in a series of reads or a random read, and we don't have access
+ // to the block index. This is costly and should happen very rarely.
headerBuf = ByteBuffer.allocate(hdrSize);
- readAtOffset(is, headerBuf.array(), headerBuf.arrayOffset(), hdrSize, false,
- offset, pread);
+ // headerBuf is HBB
+ readAtOffset(is, headerBuf.array(), headerBuf.arrayOffset(),
+ hdrSize, false, offset, pread);
}
- onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf);
- }
- int preReadHeaderSize = headerBuf == null? 0 : hdrSize;
- // Allocate enough space to fit the next block's header too; saves a seek next time through.
- // onDiskBlock is whole block + header + checksums then extra hdrSize to read next header;
- // onDiskSizeWithHeader is header, body, and any checksums if present.
- // TODO: Make this ByteBuffer-based. Will make it easier to go to HDFS with BBPool (offheap).
- byte[] onDiskBlock = new byte[onDiskSizeWithHeader + hdrSize];
- int nextBlockOnDiskSize = readAtOffset(is, onDiskBlock, preReadHeaderSize,
- onDiskSizeWithHeader - preReadHeaderSize, true, offset + preReadHeaderSize, pread);
- if (headerBuf != null) {
- // The header has been read when reading the previous block OR in a distinct header-only
- // read. Copy to this block's header.
+ // TODO: FIX!!! Expensive parse just to get a length
+ b = new HFileBlock(headerBuf, fileContext.isUseHBaseChecksum());
+ // onDiskBlock is whole block + header + checksums then extra hdrSize to read next header
+ onDiskBlock = new byte[b.getOnDiskSizeWithHeader() + hdrSize];
+ // headerBuf is HBB. Copy hdr into onDiskBlock
System.arraycopy(headerBuf.array(), headerBuf.arrayOffset(), onDiskBlock, 0, hdrSize);
- } else {
- headerBuf = ByteBuffer.wrap(onDiskBlock, 0, hdrSize);
+ nextBlockOnDiskSize = readAtOffset(is, onDiskBlock, hdrSize,
+ b.getOnDiskSizeWithHeader() - hdrSize, true, offset + hdrSize, pread);
+ onDiskSizeWithHeader = b.onDiskSizeWithoutHeader + hdrSize;
}
- // Do a few checks before we go instantiate HFileBlock.
- assert onDiskSizeWithHeader > this.hdrSize;
- verifyOnDiskSizeMatchesHeader(onDiskSizeWithHeader, headerBuf, offset);
- // The onDiskBlock will become the headerAndDataBuffer for this block.
- // If nextBlockOnDiskSizeWithHeader is not zero, the onDiskBlock already
- // contains the header of next block, so no need to set next block's header in it.
- HFileBlock hFileBlock =
- new HFileBlock(new SingleByteBuff(ByteBuffer.wrap(onDiskBlock, 0, onDiskSizeWithHeader)),
- this.fileContext.isUseHBaseChecksum(), MemoryType.EXCLUSIVE, offset,
- nextBlockOnDiskSize, fileContext);
- // Run check on uncompressed sizings.
+
if (!fileContext.isCompressedOrEncrypted()) {
- hFileBlock.sanityCheckUncompressed();
+ b.assumeUncompressed();
}
- if (verifyChecksum && !validateBlockChecksum(hFileBlock, offset, onDiskBlock, hdrSize)) {
- return null;
+
+ if (verifyChecksum && !validateBlockChecksum(b, offset, onDiskBlock, hdrSize)) {
+ return null; // checksum mismatch
}
- if (LOG.isTraceEnabled()) {
- LOG.trace("Read " + hFileBlock);
+
+ // The onDiskBlock will become the headerAndDataBuffer for this block.
+ // If nextBlockOnDiskSizeWithHeader is not zero, the onDiskBlock already
+ // contains the header of next block, so no need to set next
+ // block's header in it.
+ b = new HFileBlock(ByteBuffer.wrap(onDiskBlock, 0, onDiskSizeWithHeader),
+ this.fileContext.isUseHBaseChecksum());
+
+ b.nextBlockOnDiskSizeWithHeader = nextBlockOnDiskSize;
+
+ // Set prefetched header
+ if (b.hasNextBlockHeader()) {
+ prefetchedHeader.offset = offset + b.getOnDiskSizeWithHeader();
+ System.arraycopy(onDiskBlock, onDiskSizeWithHeader, prefetchedHeader.header, 0, hdrSize);
}
- // Cache next block header if we read it for the next time through here.
- if (nextBlockOnDiskSize != -1) {
- cacheNextBlockHeader(offset + hFileBlock.getOnDiskSizeWithHeader(),
- onDiskBlock, onDiskSizeWithHeader, hdrSize);
+
+ b.offset = offset;
+ b.fileContext.setIncludesTags(this.fileContext.isIncludesTags());
+ b.fileContext.setIncludesMvcc(this.fileContext.isIncludesMvcc());
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Read preReadHeader=" + preReadHeader + ", fastPath=" + fastPath +
+ ", readHdrOnly=" + readHdrOnly + ", " + b);
}
- return hFileBlock;
+ return b;
}
@Override
@@ -1759,73 +1819,42 @@ public class HFileBlock implements Cacheable {
}
}
- /** An additional sanity-check in case no compression or encryption is being used. */
- void sanityCheckUncompressed() throws IOException {
- if (onDiskSizeWithoutHeader != uncompressedSizeWithoutHeader +
- totalChecksumBytes()) {
- throw new IOException("Using no compression but "
- + "onDiskSizeWithoutHeader=" + onDiskSizeWithoutHeader + ", "
- + "uncompressedSizeWithoutHeader=" + uncompressedSizeWithoutHeader
- + ", numChecksumbytes=" + totalChecksumBytes());
- }
- }
-
- // Cacheable implementation
@Override
public int getSerializedLength() {
if (buf != null) {
- // Include extra bytes for block metadata.
- return this.buf.limit() + BLOCK_METADATA_SPACE;
+ // include extra bytes for the next header when it's available.
+ int extraSpace = hasNextBlockHeader() ? headerSize() : 0;
+ return this.buf.limit() + extraSpace + HFileBlock.EXTRA_SERIALIZATION_SPACE;
}
return 0;
}
- // Cacheable implementation
@Override
public void serialize(ByteBuffer destination) {
- // BE CAREFUL!! There is a custom version of this serialization over in BucketCache#doDrain.
- // Make sure any changes in here are reflected over there.
- this.buf.get(destination, 0, getSerializedLength() - BLOCK_METADATA_SPACE);
- destination = addMetaData(destination);
-
- // Make it ready for reading. flip sets position to zero and limit to current position which
- // is what we want if we do not want to serialize the block plus checksums if present plus
- // metadata.
- destination.flip();
- }
-
- /**
- * For use by bucketcache. This exposes internals.
- */
- public ByteBuffer getMetaData() {
- ByteBuffer bb = ByteBuffer.allocate(BLOCK_METADATA_SPACE);
- bb = addMetaData(bb);
- bb.flip();
- return bb;
+ this.buf.get(destination, 0, getSerializedLength() - EXTRA_SERIALIZATION_SPACE);
+ serializeExtraInfo(destination);
}
/**
- * Adds metadata at current position (position is moved forward). Does not flip or reset.
- * @return The passed <code>destination</code> with metadata added.
+ * Write out the content of EXTRA_SERIALIZATION_SPACE. Public so can be accessed by BucketCache.
*/
- private ByteBuffer addMetaData(final ByteBuffer destination) {
+ public void serializeExtraInfo(ByteBuffer destination) {
destination.put(this.fileContext.isUseHBaseChecksum() ? (byte) 1 : (byte) 0);
destination.putLong(this.offset);
- destination.putInt(this.nextBlockOnDiskSize);
- return destination;
+ destination.putInt(this.nextBlockOnDiskSizeWithHeader);
+ destination.rewind();
}
- // Cacheable implementation
@Override
public CacheableDeserializer<Cacheable> getDeserializer() {
- return HFileBlock.BLOCK_DESERIALIZER;
+ return HFileBlock.blockDeserializer;
}
@Override
public int hashCode() {
int result = 1;
result = result * 31 + blockType.hashCode();
- result = result * 31 + nextBlockOnDiskSize;
+ result = result * 31 + nextBlockOnDiskSizeWithHeader;
result = result * 31 + (int) (offset ^ (offset >>> 32));
result = result * 31 + onDiskSizeWithoutHeader;
result = result * 31 + (int) (prevBlockOffset ^ (prevBlockOffset >>> 32));
@@ -1851,10 +1880,9 @@ public class HFileBlock implements Cacheable {
if (castedComparison.blockType != this.blockType) {
return false;
}
- if (castedComparison.nextBlockOnDiskSize != this.nextBlockOnDiskSize) {
+ if (castedComparison.nextBlockOnDiskSizeWithHeader != this.nextBlockOnDiskSizeWithHeader) {
return false;
}
- // Offset is important. Needed when we have to remake cachekey when block is returned to cache.
if (castedComparison.offset != this.offset) {
return false;
}
@@ -1940,7 +1968,7 @@ public class HFileBlock implements Cacheable {
}
/**
- * @return This HFileBlocks fileContext which will a derivative of the
+ * @return the HFileContext used to create this HFileBlock. Not necessary the
* fileContext for the file from which this block's data was originally read.
*/
HFileContext getHFileContext() {
@@ -1964,7 +1992,6 @@ public class HFileBlock implements Cacheable {
* This is mostly helpful for debugging. This assumes that the block
* has minor version > 0.
*/
- @VisibleForTesting
static String toStringHeader(ByteBuff buf) throws IOException {
byte[] magicBuf = new byte[Math.min(buf.limit() - buf.position(), BlockType.MAGIC_LENGTH)];
buf.get(magicBuf);
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
index 506f08d..9f29f97 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
@@ -60,7 +60,7 @@ import org.apache.hadoop.util.StringUtils;
* Examples of how to use the block index writer can be found in
* {@link org.apache.hadoop.hbase.io.hfile.CompoundBloomFilterWriter} and
* {@link HFileWriterImpl}. Examples of how to use the reader can be
- * found in {@link HFileReaderImpl} and
+ * found in {@link HFileWriterImpl} and
* {@link org.apache.hadoop.hbase.io.hfile.TestHFileBlockIndex}.
*/
@InterfaceAudience.Private
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java
index d71911f..8f5040e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java
@@ -252,20 +252,18 @@ public class HFileReaderImpl implements HFile.Reader, Configurable {
long end = 0;
try {
end = getTrailer().getLoadOnOpenDataOffset();
+ HFileBlock prevBlock = null;
if (LOG.isTraceEnabled()) {
LOG.trace("File=" + path.toString() + ", offset=" + offset + ", end=" + end);
}
- // TODO: Could we use block iterator in here? Would that get stuff into the cache?
- HFileBlock prevBlock = null;
while (offset < end) {
if (Thread.interrupted()) {
break;
}
- // Perhaps we got our block from cache? Unlikely as this may be, if it happens, then
- // the internal-to-hfileblock thread local which holds the overread that gets the
- // next header, will not have happened...so, pass in the onDiskSize gotten from the
- // cached block. This 'optimization' triggers extremely rarely I'd say.
- long onDiskSize = prevBlock != null? prevBlock.getNextBlockOnDiskSize(): -1;
+ long onDiskSize = -1;
+ if (prevBlock != null) {
+ onDiskSize = prevBlock.getNextBlockOnDiskSizeWithHeader();
+ }
HFileBlock block = readBlock(offset, onDiskSize, true, false, false, false,
null, null);
// Need not update the current block. Ideally here the readBlock won't find the
@@ -905,8 +903,9 @@ public class HFileReaderImpl implements HFile.Reader, Configurable {
// We are reading the next block without block type validation, because
// it might turn out to be a non-data block.
- block = reader.readBlock(block.getOffset() + block.getOnDiskSizeWithHeader(),
- block.getNextBlockOnDiskSize(), cacheBlocks, pread,
+ block = reader.readBlock(block.getOffset()
+ + block.getOnDiskSizeWithHeader(),
+ block.getNextBlockOnDiskSizeWithHeader(), cacheBlocks, pread,
isCompaction, true, null, getEffectiveDataBlockEncoding());
if (block != null && !block.getBlockType().isData()) { // Findbugs: NP_NULL_ON_SOME_PATH
// Whatever block we read we will be returning it unless
@@ -1440,8 +1439,8 @@ public class HFileReaderImpl implements HFile.Reader, Configurable {
// Cache Miss, please load.
}
- HFileBlock metaBlock = fsBlockReader.readBlockData(metaBlockOffset, blockSize, true).
- unpack(hfileContext, fsBlockReader);
+ HFileBlock metaBlock = fsBlockReader.readBlockData(metaBlockOffset,
+ blockSize, -1, true).unpack(hfileContext, fsBlockReader);
// Cache the block
if (cacheBlock) {
@@ -1527,8 +1526,8 @@ public class HFileReaderImpl implements HFile.Reader, Configurable {
traceScope.getSpan().addTimelineAnnotation("blockCacheMiss");
}
// Load block from filesystem.
- HFileBlock hfileBlock =
- fsBlockReader.readBlockData(dataBlockOffset, onDiskBlockSize, pread);
+ HFileBlock hfileBlock = fsBlockReader.readBlockData(dataBlockOffset, onDiskBlockSize, -1,
+ pread);
validateBlockType(hfileBlock, expectedBlockType);
HFileBlock unpacked = hfileBlock.unpack(hfileContext, fsBlockReader);
BlockType.BlockCategory category = hfileBlock.getBlockType().getCategory();
@@ -1872,7 +1871,6 @@ public class HFileReaderImpl implements HFile.Reader, Configurable {
* @return Scanner on this file.
*/
@Override
- @VisibleForTesting
public HFileScanner getScanner(boolean cacheBlocks, final boolean pread) {
return getScanner(cacheBlocks, pread, false);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/54a543de/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java
index e0f3d74..c67bdd4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileScanner.java
@@ -99,21 +99,18 @@ public interface HFileScanner extends Shipper, Closeable {
* @throws IOException
*/
boolean seekTo() throws IOException;
-
/**
* Scans to the next entry in the file.
* @return Returns false if you are at the end otherwise true if more in file.
* @throws IOException
*/
boolean next() throws IOException;
-
/**
* Gets the current key in the form of a cell. You must call
* {@link #seekTo(Cell)} before this method.
* @return gets the current key as a Cell.
*/
Cell getKey();
-
/**
* Gets a buffer view to the current value. You must call
* {@link #seekTo(Cell)} before this method.
@@ -122,35 +119,26 @@ public interface HFileScanner extends Shipper, Closeable {
* the position is 0, the start of the buffer view.
*/
ByteBuffer getValue();
-
/**
* @return Instance of {@link org.apache.hadoop.hbase.Cell}.
*/
Cell getCell();
-
/**
* Convenience method to get a copy of the key as a string - interpreting the
* bytes as UTF8. You must call {@link #seekTo(Cell)} before this method.
* @return key as a string
- * @deprecated Since hbase-2.0.0
*/
- @Deprecated
String getKeyString();
-
/**
* Convenience method to get a copy of the value as a string - interpreting
* the bytes as UTF8. You must call {@link #seekTo(Cell)} before this method.
* @return value as a string
- * @deprecated Since hbase-2.0.0
*/
- @Deprecated
String getValueString();
-
/**
* @return Reader that underlies this Scanner instance.
*/
HFile.Reader getReader();
-
/**
* @return True is scanner has had one of the seek calls invoked; i.e.
* {@link #seekBefore(Cell)} or {@link #seekTo()} or {@link #seekTo(Cell)}.
[4/4] hbase git commit: Revert "HBASE-15392 Single Cell Get reads two
HFileBlocks M
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
moreRowsMayExistAfterCell Exploit the fact a Scan is a Get Scan. Also save
compares
Posted by st...@apache.org.
Revert "HBASE-15392 Single Cell Get reads two HFileBlocks M hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java moreRowsMayExistAfterCell Exploit the fact a Scan is a Get Scan. Also save compares if no non-default stopRow."
Revert mistaken commit
This reverts commit 7073f69993e446c2a63308f7827639d675a2da58.
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/a13d6e00
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/a13d6e00
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/a13d6e00
Branch: refs/heads/master
Commit: a13d6e000ddef2c524b641e04f7969950b2b1811
Parents: c49d0ca
Author: stack <st...@apache.org>
Authored: Tue Mar 22 18:39:22 2016 -0700
Committer: stack <st...@apache.org>
Committed: Tue Mar 22 18:39:22 2016 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/CellComparator.java | 2 +-
.../hbase/io/hfile/CombinedBlockCache.java | 13 +-
.../hadoop/hbase/io/hfile/HFileWriterImpl.java | 2 +-
.../hbase/regionserver/KeyValueScanner.java | 10 +-
.../hbase/regionserver/ScanQueryMatcher.java | 60 ++---
.../hadoop/hbase/regionserver/StoreScanner.java | 87 +------
.../hbase/util/CollectionBackedScanner.java | 3 +-
.../hbase/regionserver/KeyValueScanFixture.java | 8 +-
.../regionserver/TestKeyValueScanFixture.java | 3 +-
.../hbase/regionserver/TestStoreScanner.java | 244 +------------------
10 files changed, 62 insertions(+), 370 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-common/src/main/java/org/apache/hadoop/hbase/CellComparator.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/CellComparator.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/CellComparator.java
index 4a5c0b7..a5e26cf 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/CellComparator.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/CellComparator.java
@@ -433,7 +433,7 @@ public class CellComparator implements Comparator<Cell>, Serializable {
/**
* Used to compare two cells based on the column hint provided. This is specifically
* used when we need to optimize the seeks based on the next indexed key. This is an
- * advanced usage API specifically needed for some optimizations.
+ * advance usage API specifically needed for some optimizations.
* @param nextIndexedCell the next indexed cell
* @param currentCell the cell to be compared
* @param foff the family offset of the currentCell
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java
index 666b357..22bffee 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CombinedBlockCache.java
@@ -63,8 +63,8 @@ public class CombinedBlockCache implements ResizableBlockCache, HeapSize {
@Override
public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
final boolean cacheDataInL1) {
- boolean metaBlock = buf.getBlockType().getCategory() != BlockCategory.DATA;
- if (metaBlock || cacheDataInL1) {
+ boolean isMetaBlock = buf.getBlockType().getCategory() != BlockCategory.DATA;
+ if (isMetaBlock || cacheDataInL1) {
lruCache.cacheBlock(cacheKey, buf, inMemory, cacheDataInL1);
} else {
l2Cache.cacheBlock(cacheKey, buf, inMemory, false);
@@ -81,9 +81,12 @@ public class CombinedBlockCache implements ResizableBlockCache, HeapSize {
boolean repeat, boolean updateCacheMetrics) {
// TODO: is there a hole here, or just awkwardness since in the lruCache getBlock
// we end up calling l2Cache.getBlock.
- return lruCache.containsBlock(cacheKey)?
- lruCache.getBlock(cacheKey, caching, repeat, updateCacheMetrics):
- l2Cache.getBlock(cacheKey, caching, repeat, updateCacheMetrics);
+ if (lruCache.containsBlock(cacheKey)) {
+ return lruCache.getBlock(cacheKey, caching, repeat, updateCacheMetrics);
+ }
+ Cacheable result = l2Cache.getBlock(cacheKey, caching, repeat, updateCacheMetrics);
+
+ return result;
}
@Override
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
index 9ab46cf..d310d13 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
@@ -416,7 +416,7 @@ public class HFileWriterImpl implements HFile.Writer {
// No opportunity for optimization. Just return right key.
return right;
}
-
+
/**
* @param leftArray
* @param leftOffset
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
index e26022e..eae713f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyValueScanner.java
@@ -29,9 +29,6 @@ import org.apache.hadoop.hbase.client.Scan;
* Scanner that returns the next KeyValue.
*/
@InterfaceAudience.Private
-// TODO: Change name from KeyValueScanner to CellScanner only we already have a simple CellScanner
-// so this should be something else altogether, a decoration on our base CellScanner. TODO.
-// This class shows in CPs so do it all in one swell swoop. HBase-2.0.0.
public interface KeyValueScanner extends Shipper {
/**
* The byte array represents for NO_NEXT_INDEXED_KEY;
@@ -164,9 +161,8 @@ public interface KeyValueScanner extends Shipper {
public boolean seekToLastRow() throws IOException;
/**
- * @return the next key in the index, usually the first key of next block OR a key that falls
- * between last key of current block and first key of next block..
- * see HFileWriterImpl#getMidpoint, or null if not known.
+ * @return the next key in the index (the key to seek to the next block)
+ * if known, or null otherwise
*/
public Cell getNextIndexedKey();
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
index 706fc5b..c220b5c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanQueryMatcher.java
@@ -93,7 +93,7 @@ public class ScanQueryMatcher {
/* row is not private for tests */
/** Row the query is on */
Cell curCell;
-
+
/**
* Oldest put in any of the involved store files
* Used to decide whether it is ok to delete
@@ -119,7 +119,7 @@ public class ScanQueryMatcher {
* first column.
* */
private boolean hasNullColumn = true;
-
+
private RegionCoprocessorHost regionCoprocessorHost= null;
// By default, when hbase.hstore.time.to.purge.deletes is 0ms, a delete
@@ -140,22 +140,22 @@ public class ScanQueryMatcher {
// currently influencing. This is because Puts, that this delete can
// influence. may appear out of order.
private final long timeToPurgeDeletes;
-
+
private final boolean isUserScan;
private final boolean isReversed;
/**
- * True if we are doing a 'Get' Scan. Every Get is actually a one-row Scan.
- */
- private final boolean get;
-
- /**
* Construct a QueryMatcher for a scan
+ * @param scan
* @param scanInfo The store's immutable scan info
+ * @param columns
* @param scanType Type of the scan
* @param earliestPutTs Earliest put seen in any of the store files.
- * @param oldestUnexpiredTS the oldest timestamp we are interested in, based on TTL
+ * @param oldestUnexpiredTS the oldest timestamp we are interested in,
+ * based on TTL
+ * @param regionCoprocessorHost
+ * @throws IOException
*/
public ScanQueryMatcher(Scan scan, ScanInfo scanInfo, NavigableSet<byte[]> columns,
ScanType scanType, long readPointToUse, long earliestPutTs, long oldestUnexpiredTS,
@@ -166,7 +166,6 @@ public class ScanQueryMatcher {
} else {
this.tr = timeRange;
}
- this.get = scan.isGetScan();
this.rowComparator = scanInfo.getComparator();
this.regionCoprocessorHost = regionCoprocessorHost;
this.deletes = instantiateDeleteTracker();
@@ -235,8 +234,8 @@ public class ScanQueryMatcher {
* @param now the current server time
* @param dropDeletesFromRow The inclusive left bound of the range; can be EMPTY_START_ROW.
* @param dropDeletesToRow The exclusive right bound of the range; can be EMPTY_END_ROW.
- * @param regionCoprocessorHost
- * @throws IOException
+ * @param regionCoprocessorHost
+ * @throws IOException
*/
public ScanQueryMatcher(Scan scan, ScanInfo scanInfo, NavigableSet<byte[]> columns,
long readPointToUse, long earliestPutTs, long oldestUnexpiredTS, long now,
@@ -281,7 +280,7 @@ public class ScanQueryMatcher {
* caused by a data corruption.
*/
public MatchCode match(Cell cell) throws IOException {
- if (filter != null && filter.filterAllRemaining()) {
+ if (filter != null && filter.filterAllRemaining()) {
return MatchCode.DONE_SCAN;
}
if (curCell != null) {
@@ -325,7 +324,7 @@ public class ScanQueryMatcher {
// check if the cell is expired by cell TTL
if (HStore.isCellTTLExpired(cell, this.oldestUnexpiredTS, this.now)) {
return MatchCode.SKIP;
- }
+ }
/*
* The delete logic is pretty complicated now.
@@ -360,10 +359,10 @@ public class ScanQueryMatcher {
}
// Can't early out now, because DelFam come before any other keys
}
-
+
if ((!isUserScan)
&& timeToPurgeDeletes > 0
- && (EnvironmentEdgeManager.currentTime() - timestamp)
+ && (EnvironmentEdgeManager.currentTime() - timestamp)
<= timeToPurgeDeletes) {
return MatchCode.INCLUDE;
} else if (retainDeletesInOutput || mvccVersion > maxReadPointToTrackVersions) {
@@ -504,27 +503,22 @@ public class ScanQueryMatcher {
}
}
- /**
- * @return Returns false if we know there are no more rows to be scanned (We've reached the
- * <code>stopRow</code> or we are scanning on row only because this Scan is for a Get, etc.
- */
public boolean moreRowsMayExistAfter(Cell kv) {
- // If a 'get' Scan -- we are doing a Get (every Get is a single-row Scan in implementation) --
- // then we are looking at one row only, the one specified in the Get coordinate..so we know
- // for sure that there are no more rows on this Scan
- if (this.get) {
- return false;
+ if (this.isReversed) {
+ if (rowComparator.compareRows(kv, stopRow, 0, stopRow.length) <= 0) {
+ return false;
+ } else {
+ return true;
+ }
}
- // If no stopRow, return that there may be more rows. The tests that follow depend on a
- // non-empty, non-default stopRow so this little test below short-circuits out doing the
- // following compares.
- if (this.stopRow == null || this.stopRow == HConstants.EMPTY_BYTE_ARRAY) {
+ if (!Bytes.equals(stopRow , HConstants.EMPTY_END_ROW) &&
+ rowComparator.compareRows(kv, stopRow, 0, stopRow.length) >= 0) {
+ // KV >= STOPROW
+ // then NO there is nothing left.
+ return false;
+ } else {
return true;
}
- return this.isReversed?
- rowComparator.compareRows(kv, stopRow, 0, stopRow.length) > 0:
- Bytes.equals(stopRow, HConstants.EMPTY_END_ROW) ||
- rowComparator.compareRows(kv, stopRow, 0, stopRow.length) < 0;
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index 66f846a..2f0d284 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -133,7 +133,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
protected List<KeyValueScanner> currentScanners = new ArrayList<KeyValueScanner>();
// flush update lock
private ReentrantLock flushLock = new ReentrantLock();
-
+
protected final long readPt;
// used by the injection framework to test race between StoreScanner construction and compaction
@@ -600,12 +600,6 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
continue;
case DONE:
- // Optimization for Gets! If DONE, no more to get on this row, early exit!
- if (this.scan.isGetScan()) {
- // Then no more to this row... exit.
- close(false);// Do all cleanup except heap.close()
- return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
- }
matcher.curCell = null;
return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
@@ -655,67 +649,18 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
}
- /**
- * See if we should actually SEEK or rather just SKIP to the next Cell (see HBASE-13109).
- * This method works together with ColumnTrackers and Filters. ColumnTrackers may issue SEEK
- * hints, such as seek to next column, next row, or seek to an arbitrary seek key.
- * This method intercepts these qcodes and decides whether a seek is the most efficient _actual_
- * way to get us to the requested cell (SEEKs are more expensive than SKIP, SKIP, SKIP inside the
- * current, loaded block).
- * It does this by looking at the next indexed key of the current HFile. This key
- * is then compared with the _SEEK_ key, where a SEEK key is an artificial 'last possible key
- * on the row' (only in here, we avoid actually creating a SEEK key; in the compare we work with
- * the current Cell but compare as though it were a seek key; see down in
- * matcher.compareKeyForNextRow, etc). If the compare gets us onto the
- * next block we *_SEEK, otherwise we just INCLUDE or SKIP, and let the ColumnTrackers or Filters
- * go through the next Cell, and so on)
- *
- * <p>The ColumnTrackers and Filters must behave correctly in all cases, i.e. if they are past the
- * Cells they care about they must issues a SKIP or SEEK.
- *
- * <p>Other notes:
- * <ul>
- * <li>Rows can straddle block boundaries</li>
- * <li>Versions of columns can straddle block boundaries (i.e. column C1 at T1 might be in a
- * different block than column C1 at T2)</li>
- * <li>We want to SKIP and INCLUDE if the chance is high that we'll find the desired Cell after a
- * few SKIPs...</li>
- * <li>We want to INCLUDE_AND_SEEK and SEEK when the chance is high that we'll be able to seek
- * past many Cells, especially if we know we need to go to the next block.</li>
- * </ul>
- * <p>A good proxy (best effort) to determine whether INCLUDE/SKIP is better than SEEK is whether
- * we'll likely end up seeking to the next block (or past the next block) to get our next column.
- * Example:
- * <pre>
- * | BLOCK 1 | BLOCK 2 |
- * | r1/c1, r1/c2, r1/c3 | r1/c4, r1/c5, r2/c1 |
- * ^ ^
- * | |
- * Next Index Key SEEK_NEXT_ROW (before r2/c1)
- *
- *
- * | BLOCK 1 | BLOCK 2 |
- * | r1/c1/t5, r1/c1/t4, r1/c1/t3 | r1/c1/t2, r1/c1/T1, r1/c2/T3 |
- * ^ ^
- * | |
- * Next Index Key SEEK_NEXT_COL
- * </pre>
- * Now imagine we want columns c1 and c3 (see first diagram above), the 'Next Index Key' of r1/c4
- * is > r1/c3 so we should seek to get to the c1 on the next row, r2. In second case, say we only
- * want one version of c1, after we have it, a SEEK_COL will be issued to get to c2. Looking at
- * the 'Next Index Key', it would land us in the next block, so we should SEEK. In other scenarios
- * where the SEEK will not land us in the next block, it is very likely better to issues a series
- * of SKIPs.
+ /*
+ * See if we should actually SEEK or rather just SKIP to the next Cell.
+ * (see HBASE-13109)
*/
- @VisibleForTesting
- protected ScanQueryMatcher.MatchCode optimize(ScanQueryMatcher.MatchCode qcode, Cell cell) {
+ private ScanQueryMatcher.MatchCode optimize(ScanQueryMatcher.MatchCode qcode, Cell cell) {
switch(qcode) {
case INCLUDE_AND_SEEK_NEXT_COL:
case SEEK_NEXT_COL:
{
Cell nextIndexedKey = getNextIndexedKey();
if (nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY
- && matcher.compareKeyForNextColumn(nextIndexedKey, cell) > 0) {
+ && matcher.compareKeyForNextColumn(nextIndexedKey, cell) >= 0) {
return qcode == MatchCode.SEEK_NEXT_COL ? MatchCode.SKIP : MatchCode.INCLUDE;
}
break;
@@ -723,16 +668,10 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
case INCLUDE_AND_SEEK_NEXT_ROW:
case SEEK_NEXT_ROW:
{
- // If it is a Get Scan, then we know that we are done with this row; there are no more
- // rows beyond the current one: don't try to optimize. We are DONE. Return the *_NEXT_ROW
- // qcode as is. When the caller gets these flags on a Get Scan, it knows it can shut down the
- // Scan.
- if (!this.scan.isGetScan()) {
- Cell nextIndexedKey = getNextIndexedKey();
- if (nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY
- && matcher.compareKeyForNextRow(nextIndexedKey, cell) > 0) {
- return qcode == MatchCode.SEEK_NEXT_ROW ? MatchCode.SKIP : MatchCode.INCLUDE;
- }
+ Cell nextIndexedKey = getNextIndexedKey();
+ if (nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY
+ && matcher.compareKeyForNextRow(nextIndexedKey, cell) >= 0) {
+ return qcode == MatchCode.SEEK_NEXT_ROW ? MatchCode.SKIP : MatchCode.INCLUDE;
}
break;
}
@@ -870,10 +809,10 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
// check the var without any lock. Suppose even if we see the old
// value here still it is ok to continue because we will not be resetting
// the heap but will continue with the referenced memstore's snapshot. For compactions
- // any way we don't need the updateReaders at all to happen as we still continue with
+ // any way we don't need the updateReaders at all to happen as we still continue with
// the older files
if (flushed) {
- // If there is a flush and the current scan is notified on the flush ensure that the
+ // If there is a flush and the current scan is notified on the flush ensure that the
// scan's heap gets reset and we do a seek on the newly flushed file.
if(!this.closing) {
this.lastTop = this.peek();
@@ -903,7 +842,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
if (scanners.isEmpty()) return;
int storeFileScannerCount = scanners.size();
CountDownLatch latch = new CountDownLatch(storeFileScannerCount);
- List<ParallelSeekHandler> handlers =
+ List<ParallelSeekHandler> handlers =
new ArrayList<ParallelSeekHandler>(storeFileScannerCount);
for (KeyValueScanner scanner : scanners) {
if (scanner instanceof StoreFileScanner) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java
index 4720880..9fc068f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/CollectionBackedScanner.java
@@ -30,7 +30,8 @@ import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.regionserver.NonReversedNonLazyKeyValueScanner;
/**
- * Utility scanner that wraps a sortable collection and serves as a KeyValueScanner.
+ * Utility scanner that wraps a sortable collection and serves
+ * as a KeyValueScanner.
*/
@InterfaceAudience.Private
public class CollectionBackedScanner extends NonReversedNonLazyKeyValueScanner {
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/KeyValueScanFixture.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/KeyValueScanFixture.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/KeyValueScanFixture.java
index a4e7f9b..3f87a00 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/KeyValueScanFixture.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/KeyValueScanFixture.java
@@ -22,7 +22,6 @@ package org.apache.hadoop.hbase.regionserver;
import java.util.ArrayList;
import java.util.List;
-import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.util.CollectionBackedScanner;
@@ -34,8 +33,9 @@ import org.apache.hadoop.hbase.util.CollectionBackedScanner;
* to be a store file scanner.
*/
public class KeyValueScanFixture extends CollectionBackedScanner {
- public KeyValueScanFixture(CellComparator comparator, Cell... cells) {
- super(comparator, cells);
+ public KeyValueScanFixture(CellComparator comparator,
+ KeyValue... incData) {
+ super(comparator, incData);
}
public static List<KeyValueScanner> scanFixture(KeyValue[] ... kvArrays) {
@@ -45,4 +45,4 @@ public class KeyValueScanFixture extends CollectionBackedScanner {
}
return scanners;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestKeyValueScanFixture.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestKeyValueScanFixture.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestKeyValueScanFixture.java
index 0e96682..a8c2c65 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestKeyValueScanFixture.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestKeyValueScanFixture.java
@@ -46,7 +46,8 @@ public class TestKeyValueScanFixture extends TestCase {
KeyValueTestUtil.create("RowB", "family", "qf1",
10, KeyValue.Type.Put, "value-10")
};
- KeyValueScanner scan = new KeyValueScanFixture(CellComparator.COMPARATOR, kvs);
+ KeyValueScanner scan = new KeyValueScanFixture(
+ CellComparator.COMPARATOR, kvs);
KeyValue kv = KeyValueUtil.createFirstOnRow(Bytes.toBytes("RowA"));
// should seek to this:
http://git-wip-us.apache.org/repos/asf/hbase/blob/a13d6e00/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
index 4c594b0..92c85aa 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
@@ -20,7 +20,6 @@
package org.apache.hadoop.hbase.regionserver;
import static org.apache.hadoop.hbase.regionserver.KeyValueScanFixture.scanFixture;
-import static org.junit.Assert.*;
import java.io.IOException;
import java.util.ArrayList;
@@ -28,21 +27,16 @@ import java.util.Arrays;
import java.util.List;
import java.util.NavigableSet;
import java.util.TreeSet;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CategoryBasedTimeout;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
-import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeepDeletedCells;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueTestUtil;
-import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests;
@@ -59,113 +53,16 @@ import org.junit.rules.TestRule;
// Can't be small as it plays with EnvironmentEdgeManager
@Category({RegionServerTests.class, MediumTests.class})
public class TestStoreScanner {
- private static final Log LOG = LogFactory.getLog(TestStoreScanner.class);
@Rule public TestName name = new TestName();
@Rule public final TestRule timeout = CategoryBasedTimeout.builder().withTimeout(this.getClass()).
withLookingForStuckThread(true).build();
private static final String CF_STR = "cf";
- private static final byte [] CF = Bytes.toBytes(CF_STR);
+ final byte [] CF = Bytes.toBytes(CF_STR);
static Configuration CONF = HBaseConfiguration.create();
private ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, Integer.MAX_VALUE,
Long.MAX_VALUE, KeepDeletedCells.FALSE, 0, CellComparator.COMPARATOR);
private ScanType scanType = ScanType.USER_SCAN;
- /**
- * From here on down, we have a bunch of defines and specific CELL_GRID of Cells. The
- * CELL_GRID then has a Scanner that can fake out 'block' transitions. All this elaborate
- * setup is for tests that ensure we don't overread, and that the
- * {@link StoreScanner#optimize(org.apache.hadoop.hbase.regionserver.ScanQueryMatcher.MatchCode,
- * Cell)} is not overly enthusiastic.
- */
- private static final byte [] ZERO = new byte [] {'0'};
- private static final byte [] ZERO_POINT_ZERO = new byte [] {'0', '.', '0'};
- private static final byte [] ONE = new byte [] {'1'};
- private static final byte [] TWO = new byte [] {'2'};
- private static final byte [] TWO_POINT_TWO = new byte [] {'2', '.', '2'};
- private static final byte [] THREE = new byte [] {'3'};
- private static final byte [] FOUR = new byte [] {'4'};
- private static final byte [] FIVE = new byte [] {'5'};
- private static final byte [] VALUE = new byte [] {'v'};
- private static final int CELL_GRID_BLOCK2_BOUNDARY = 4;
- private static final int CELL_GRID_BLOCK3_BOUNDARY = 11;
- private static final int CELL_GRID_BLOCK4_BOUNDARY = 15;
- private static final int CELL_GRID_BLOCK5_BOUNDARY = 19;
-
- /**
- * Five rows by four columns distinguished by column qualifier (column qualifier is one of the
- * four rows... ONE, TWO, etc.). Exceptions are a weird row after TWO; it is TWO_POINT_TWO.
- * And then row FOUR has five columns finishing w/ row FIVE having a single column.
- * We will use this to test scan does the right thing as it
- * we do Gets, StoreScanner#optimize, and what we do on (faked) block boundaries.
- */
- private static final Cell [] CELL_GRID = new Cell [] {
- CellUtil.createCell(ONE, CF, ONE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(ONE, CF, TWO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(ONE, CF, THREE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(ONE, CF, FOUR, 1L, KeyValue.Type.Put.getCode(), VALUE),
- // Offset 4 CELL_GRID_BLOCK2_BOUNDARY
- CellUtil.createCell(TWO, CF, ONE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(TWO, CF, TWO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(TWO, CF, THREE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(TWO, CF, FOUR, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(TWO_POINT_TWO, CF, ZERO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(TWO_POINT_TWO, CF, ZERO_POINT_ZERO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(TWO_POINT_TWO, CF, FIVE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- // Offset 11! CELL_GRID_BLOCK3_BOUNDARY
- CellUtil.createCell(THREE, CF, ONE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(THREE, CF, TWO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(THREE, CF, THREE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(THREE, CF, FOUR, 1L, KeyValue.Type.Put.getCode(), VALUE),
- // Offset 15 CELL_GRID_BLOCK4_BOUNDARY
- CellUtil.createCell(FOUR, CF, ONE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(FOUR, CF, TWO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(FOUR, CF, THREE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(FOUR, CF, FOUR, 1L, KeyValue.Type.Put.getCode(), VALUE),
- // Offset 19 CELL_GRID_BLOCK5_BOUNDARY
- CellUtil.createCell(FOUR, CF, FIVE, 1L, KeyValue.Type.Put.getCode(), VALUE),
- CellUtil.createCell(FIVE, CF, ZERO, 1L, KeyValue.Type.Put.getCode(), VALUE),
- };
-
- /**
- * A StoreScanner for our CELL_GRID above. Fakes the block transitions. Does counts of
- * calls to optimize and counts of when optimize actually did an optimize.
- */
- private static class CellGridStoreScanner extends StoreScanner {
- // Count of how often optimize is called and of how often it does an optimize.
- final AtomicInteger count = new AtomicInteger(0);
- final AtomicInteger optimization = new AtomicInteger(0);
-
- CellGridStoreScanner(final Scan scan, ScanInfo scanInfo, ScanType scanType)
- throws IOException {
- super(scan, scanInfo, scanType, scan.getFamilyMap().get(CF),
- Arrays.<KeyValueScanner>asList(
- new KeyValueScanner[] {new KeyValueScanFixture(CellComparator.COMPARATOR, CELL_GRID)}));
- }
-
- protected ScanQueryMatcher.MatchCode optimize(ScanQueryMatcher.MatchCode qcode, Cell cell) {
- count.incrementAndGet();
- ScanQueryMatcher.MatchCode after = super.optimize(qcode, cell);
- LOG.info("Cell=" + cell + ", nextIndex=" + CellUtil.toString(getNextIndexedKey(), false) +
- ", before=" + qcode + ", after=" + after);
- if (qcode != after) {
- optimization.incrementAndGet();
- }
- return after;
- };
-
- @Override
- public Cell getNextIndexedKey() {
- // Fake block boundaries by having index of next block change as we go through scan.
- return count.get() > CELL_GRID_BLOCK4_BOUNDARY?
- CellUtil.createFirstOnRow(CELL_GRID[CELL_GRID_BLOCK5_BOUNDARY]):
- count.get() > CELL_GRID_BLOCK3_BOUNDARY?
- CellUtil.createFirstOnRow(CELL_GRID[CELL_GRID_BLOCK4_BOUNDARY]):
- count.get() > CELL_GRID_BLOCK2_BOUNDARY?
- CellUtil.createFirstOnRow(CELL_GRID[CELL_GRID_BLOCK3_BOUNDARY]):
- CellUtil.createFirstOnRow(CELL_GRID[CELL_GRID_BLOCK2_BOUNDARY]);
- }
- };
-
/*
* Test utility for building a NavigableSet for scanners.
* @param strCols
@@ -181,145 +78,6 @@ public class TestStoreScanner {
}
@Test
- public void testFullRowGetDoesNotOverreadWhenRowInsideOneBlock() throws IOException {
- // Do a Get against row two. Row two is inside a block that starts with row TWO but ends with
- // row TWO_POINT_TWO. We should read one block only.
- Get get = new Get(TWO);
- Scan scan = new Scan(get);
- CellGridStoreScanner scanner = new CellGridStoreScanner(scan, this.scanInfo, this.scanType);
- try {
- List<Cell> results = new ArrayList<Cell>();
- while (scanner.next(results)) {
- continue;
- }
- // Should be four results of column 1 (though there are 5 rows in the CELL_GRID -- the
- // TWO_POINT_TWO row does not have a a column ONE.
- Assert.assertEquals(4, results.size());
- // We should have gone the optimize route 5 times totally... an INCLUDE for the four cells
- // in the row plus the DONE on the end.
- Assert.assertEquals(5, scanner.count.get());
- // For a full row Get, there should be no opportunity for scanner optimization.
- Assert.assertEquals(0, scanner.optimization.get());
- } finally {
- scanner.close();
- }
- }
-
- @Test
- public void testFullRowSpansBlocks() throws IOException {
- // Do a Get against row FOUR. It spans two blocks.
- Get get = new Get(FOUR);
- Scan scan = new Scan(get);
- CellGridStoreScanner scanner = new CellGridStoreScanner(scan, this.scanInfo, this.scanType);
- try {
- List<Cell> results = new ArrayList<Cell>();
- while (scanner.next(results)) {
- continue;
- }
- // Should be four results of column 1 (though there are 5 rows in the CELL_GRID -- the
- // TWO_POINT_TWO row does not have a a column ONE.
- Assert.assertEquals(5, results.size());
- // We should have gone the optimize route 6 times totally... an INCLUDE for the five cells
- // in the row plus the DONE on the end.
- Assert.assertEquals(6, scanner.count.get());
- // For a full row Get, there should be no opportunity for scanner optimization.
- Assert.assertEquals(0, scanner.optimization.get());
- } finally {
- scanner.close();
- }
- }
-
- /**
- * Test optimize in StoreScanner. Test that we skip to the next 'block' when we it makes sense
- * reading the block 'index'.
- * @throws IOException
- */
- @Test
- public void testOptimize() throws IOException {
- Scan scan = new Scan();
- // A scan that just gets the first qualifier on each row of the CELL_GRID
- scan.addColumn(CF, ONE);
- CellGridStoreScanner scanner = new CellGridStoreScanner(scan, this.scanInfo, this.scanType);
- try {
- List<Cell> results = new ArrayList<Cell>();
- while (scanner.next(results)) {
- continue;
- }
- // Should be four results of column 1 (though there are 5 rows in the CELL_GRID -- the
- // TWO_POINT_TWO row does not have a a column ONE.
- Assert.assertEquals(4, results.size());
- for (Cell cell: results) {
- assertTrue(Bytes.equals(ONE, 0, ONE.length,
- cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
- }
- Assert.assertTrue("Optimize should do some optimizations", scanner.optimization.get() > 0);
- } finally {
- scanner.close();
- }
- }
-
- /**
- * Ensure the optimize Scan method in StoreScanner does not get in the way of a Get doing minimum
- * work... seeking to start of block and then SKIPPING until we find the wanted Cell.
- * This 'simple' scenario mimics case of all Cells fitting inside a single HFileBlock.
- * See HBASE-15392. This test is a little cryptic. Takes a bit of staring to figure what it up to.
- * @throws IOException
- */
- @Test
- public void testOptimizeAndGet() throws IOException {
- // First test a Get of two columns in the row R2. Every Get is a Scan. Get columns named
- // R2 and R3.
- Get get = new Get(TWO);
- get.addColumn(CF, TWO);
- get.addColumn(CF, THREE);
- Scan scan = new Scan(get);
- CellGridStoreScanner scanner = new CellGridStoreScanner(scan, this.scanInfo, this.scanType);
- try {
- List<Cell> results = new ArrayList<Cell>();
- // For a Get there should be no more next's after the first call.
- Assert.assertEquals(false, scanner.next(results));
- // Should be one result only.
- Assert.assertEquals(2, results.size());
- // And we should have gone through optimize twice only.
- Assert.assertEquals("First qcode is SEEK_NEXT_COL and second INCLUDE_AND_SEEK_NEXT_ROW",
- 3, scanner.count.get());
- } finally {
- scanner.close();
- }
- }
-
- /**
- * Ensure that optimize does not cause the Get to do more seeking than required. Optimize
- * (see HBASE-15392) was causing us to seek all Cells in a block when a Get Scan if the next block
- * index/start key was a different row to the current one. A bug. We'd call next too often
- * because we had to exhaust all Cells in the current row making us load the next block just to
- * discard what we read there. This test is a little cryptic. Takes a bit of staring to figure
- * what it up to.
- * @throws IOException
- */
- @Test
- public void testOptimizeAndGetWithFakedNextBlockIndexStart() throws IOException {
- // First test a Get of second column in the row R2. Every Get is a Scan. Second column has a
- // qualifier of R2.
- Get get = new Get(THREE);
- get.addColumn(CF, TWO);
- Scan scan = new Scan(get);
- CellGridStoreScanner scanner = new CellGridStoreScanner(scan, this.scanInfo, this.scanType);
- try {
- List<Cell> results = new ArrayList<Cell>();
- // For a Get there should be no more next's after the first call.
- Assert.assertEquals(false, scanner.next(results));
- // Should be one result only.
- Assert.assertEquals(1, results.size());
- // And we should have gone through optimize twice only.
- Assert.assertEquals("First qcode is SEEK_NEXT_COL and second INCLUDE_AND_SEEK_NEXT_ROW",
- 2, scanner.count.get());
- } finally {
- scanner.close();
- }
- }
-
- @Test
public void testScanTimeRange() throws IOException {
String r1 = "R1";
// returns only 1 of these 2 even though same timestamp