From 5b319752dea879071ee3a23ed8faba4891977f62 Mon Sep 17 00:00:00 2001 From: "Tak Lon (Stephen) Wu" Date: Mon, 16 May 2022 11:21:25 -0700 Subject: [PATCH] HBASE-27033 Backport "HBASE-27013 Introduce read all bytes when using pread for prefetch" (#4429) - introduce optional flag `hfile.pread.all.bytes.enabled` for pread that must read full bytes with the next block header Signed-off-by: Ankit Singhal Signed-off-by: Josh Elser --- .../org/apache/hadoop/hbase/HConstants.java | 7 ++ .../hadoop/hbase/io/util/BlockIOUtils.java | 23 +++- .../hadoop/hbase/io/hfile/HFileBlock.java | 7 +- .../hadoop/hbase/io/hfile/ReaderContext.java | 8 ++ .../hbase/io/hfile/TestBlockIOUtils.java | 114 ++++++++++++++++++ 5 files changed, 157 insertions(+), 2 deletions(-) diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java index 4b2eaa0bd983..3766da718d5b 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java @@ -1037,6 +1037,13 @@ public enum OperationStatusCode { public static final long HBASE_CLIENT_SCANNER_ONHEAP_BLOCK_CACHE_FIXED_SIZE_DEFAULT = 32 * 1024 * 1024L; + /** + * Configuration key for setting pread must read both necessaryLen and extraLen, default is + * disabled. This is an optimized flag for reading HFile from blob storage. + */ + public static final String HFILE_PREAD_ALL_BYTES_ENABLED_KEY = "hfile.pread.all.bytes.enabled"; + public static final boolean HFILE_PREAD_ALL_BYTES_ENABLED_DEFAULT = false; + /* * Minimum percentage of free heap necessary for a successful cluster startup. */ diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java index 1e70d2d3dafd..e24e9f87f824 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java @@ -205,10 +205,31 @@ public static boolean readWithExtra(ByteBuff buf, FSDataInputStream dis, int nec */ public static boolean preadWithExtra(ByteBuff buff, FSDataInputStream dis, long position, int necessaryLen, int extraLen) throws IOException { + return preadWithExtra(buff, dis, position, necessaryLen, extraLen, false); + } + + /** + * Read from an input stream at least necessaryLen and if possible, + * extraLen also if available. Analogous to + * {@link IOUtils#readFully(InputStream, byte[], int, int)}, but uses positional read and + * specifies a number of "extra" bytes that would be desirable but not absolutely necessary to + * read. + * @param buff ByteBuff to read into. + * @param dis the input stream to read from + * @param position the position within the stream from which to start reading + * @param necessaryLen the number of bytes that are absolutely necessary to read + * @param extraLen the number of extra bytes that would be nice to read + * @param readAllBytes whether we must read the necessaryLen and extraLen + * @return true if and only if extraLen is > 0 and reading those extra bytes was successful + * @throws IOException if failed to read the necessary bytes + */ + public static boolean preadWithExtra(ByteBuff buff, FSDataInputStream dis, long position, + int necessaryLen, int extraLen, boolean readAllBytes) throws IOException { int remain = necessaryLen + extraLen; byte[] buf = new byte[remain]; int bytesRead = 0; - while (bytesRead < necessaryLen) { + int lengthMustRead = readAllBytes ? remain : necessaryLen; + while (bytesRead < lengthMustRead) { int ret = dis.read(position + bytesRead, buf, bytesRead, remain); if (ret < 0) { throw new IOException("Premature EOF from inputStream (positional read returned " + ret 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 505e00d893c5..c08006a7ff7d 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 @@ -1345,6 +1345,8 @@ static class FSReaderImpl implements FSReader { private final Lock streamLock = new ReentrantLock(); + private final boolean isPreadAllBytes; + FSReaderImpl(ReaderContext readerContext, HFileContext fileContext, ByteBuffAllocator allocator) throws IOException { this.fileSize = readerContext.getFileSize(); @@ -1361,6 +1363,7 @@ static class FSReaderImpl implements FSReader { this.streamWrapper.prepareForBlockReader(!fileContext.isUseHBaseChecksum()); defaultDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext); encodedBlockDecodingCtx = defaultDecodingCtx; + isPreadAllBytes = readerContext.isPreadAllBytes(); } @Override @@ -1449,7 +1452,9 @@ protected boolean readAtOffset(FSDataInputStream istream, ByteBuff dest, int siz } else { // Positional read. Better for random reads; or when the streamLock is already locked. int extraSize = peekIntoNextBlock ? hdrSize : 0; - if (!BlockIOUtils.preadWithExtra(dest, istream, fileOffset, size, extraSize)) { + if ( + !BlockIOUtils.preadWithExtra(dest, istream, fileOffset, size, extraSize, isPreadAllBytes) + ) { // did not read the next block header. return false; } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ReaderContext.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ReaderContext.java index c652c4a18b53..d6f711d866eb 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ReaderContext.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/ReaderContext.java @@ -18,6 +18,7 @@ package org.apache.hadoop.hbase.io.hfile; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.fs.HFileSystem; import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper; import org.apache.yetus.audience.InterfaceAudience; @@ -39,6 +40,7 @@ public enum ReaderType { private final HFileSystem hfs; private final boolean primaryReplicaReader; private final ReaderType type; + private final boolean preadAllBytes; public ReaderContext(Path filePath, FSDataInputStreamWrapper fsdis, long fileSize, HFileSystem hfs, boolean primaryReplicaReader, ReaderType type) { @@ -48,6 +50,8 @@ public ReaderContext(Path filePath, FSDataInputStreamWrapper fsdis, long fileSiz this.hfs = hfs; this.primaryReplicaReader = primaryReplicaReader; this.type = type; + this.preadAllBytes = hfs.getConf().getBoolean(HConstants.HFILE_PREAD_ALL_BYTES_ENABLED_KEY, + HConstants.HFILE_PREAD_ALL_BYTES_ENABLED_DEFAULT); } public Path getFilePath() { @@ -73,4 +77,8 @@ public boolean isPrimaryReplicaReader() { public ReaderType getReaderType() { return this.type; } + + public boolean isPreadAllBytes() { + return preadAllBytes; + } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java index 00a38ee22729..01cb9cb7e173 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java @@ -18,6 +18,7 @@ package org.apache.hadoop.hbase.io.hfile; import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -26,14 +27,22 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; +import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Random; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseClassTestRule; import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.fs.HFileSystem; +import org.apache.hadoop.hbase.io.ByteBuffAllocator; +import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper; +import org.apache.hadoop.hbase.io.compress.Compression; import org.apache.hadoop.hbase.io.util.BlockIOUtils; import org.apache.hadoop.hbase.nio.ByteBuff; import org.apache.hadoop.hbase.nio.MultiByteBuff; @@ -41,11 +50,13 @@ 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.EnvironmentEdgeManager; import org.junit.ClassRule; import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; +import org.junit.rules.TestName; @Category({ IOTests.class, SmallTests.class }) public class TestBlockIOUtils { @@ -54,11 +65,17 @@ public class TestBlockIOUtils { public static final HBaseClassTestRule CLASS_RULE = HBaseClassTestRule.forClass(TestBlockIOUtils.class); + @Rule + public TestName testName = new TestName(); + @Rule public ExpectedException exception = ExpectedException.none(); private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); + private static final int NUM_TEST_BLOCKS = 2; + private static final Compression.Algorithm COMPRESSION_ALGO = Compression.Algorithm.GZ; + @Test public void testIsByteBufferReadable() throws IOException { FileSystem fs = TEST_UTIL.getTestFileSystem(); @@ -89,6 +106,103 @@ public void testReadFully() throws IOException { assertArrayEquals(Bytes.toBytes(s), heapBuf); } + @Test + public void testPreadWithReadFullBytes() throws IOException { + testPreadReadFullBytesInternal(true, EnvironmentEdgeManager.currentTime()); + } + + @Test + public void testPreadWithoutReadFullBytes() throws IOException { + testPreadReadFullBytesInternal(false, EnvironmentEdgeManager.currentTime()); + } + + private void testPreadReadFullBytesInternal(boolean readAllBytes, long randomSeed) + throws IOException { + Configuration conf = TEST_UTIL.getConfiguration(); + conf.setBoolean(HConstants.HFILE_PREAD_ALL_BYTES_ENABLED_KEY, readAllBytes); + FileSystem fs = TEST_UTIL.getTestFileSystem(); + Path path = new Path(TEST_UTIL.getDataTestDirOnTestFS(), testName.getMethodName()); + // give a fixed seed such we can see failure easily. + Random rand = new Random(randomSeed); + long totalDataBlockBytes = + writeBlocks(TEST_UTIL.getConfiguration(), rand, COMPRESSION_ALGO, path); + readDataBlocksAndVerify(fs, path, COMPRESSION_ALGO, totalDataBlockBytes); + } + + private long writeBlocks(Configuration conf, Random rand, Compression.Algorithm compressAlgo, + Path path) throws IOException { + FileSystem fs = HFileSystem.get(conf); + FSDataOutputStream os = fs.create(path); + HFileContext meta = + new HFileContextBuilder().withHBaseCheckSum(true).withCompression(compressAlgo).build(); + HFileBlock.Writer hbw = new HFileBlock.Writer(null, meta); + long totalDataBlockBytes = 0; + for (int i = 0; i < NUM_TEST_BLOCKS; ++i) { + int blockTypeOrdinal = rand.nextInt(BlockType.values().length); + if (blockTypeOrdinal == BlockType.ENCODED_DATA.ordinal()) { + blockTypeOrdinal = BlockType.DATA.ordinal(); + } + BlockType bt = BlockType.values()[blockTypeOrdinal]; + DataOutputStream dos = hbw.startWriting(bt); + int size = rand.nextInt(500); + for (int j = 0; j < size; ++j) { + dos.writeShort(i + 1); + dos.writeInt(j + 1); + } + + hbw.writeHeaderAndData(os); + totalDataBlockBytes += hbw.getOnDiskSizeWithHeader(); + } + // append a dummy trailer and in a actual HFile it should have more data. + FixedFileTrailer trailer = new FixedFileTrailer(3, 3); + trailer.setFirstDataBlockOffset(0); + trailer.setLastDataBlockOffset(totalDataBlockBytes); + trailer.setComparatorClass(meta.getCellComparator().getClass()); + trailer.setDataIndexCount(NUM_TEST_BLOCKS); + trailer.setCompressionCodec(compressAlgo); + trailer.serialize(os); + // close the stream + os.close(); + return totalDataBlockBytes; + } + + private void readDataBlocksAndVerify(FileSystem fs, Path path, Compression.Algorithm compressAlgo, + long totalDataBlockBytes) throws IOException { + FSDataInputStream is = fs.open(path); + HFileContext fileContext = + new HFileContextBuilder().withHBaseCheckSum(true).withCompression(compressAlgo).build(); + ReaderContext context = + new ReaderContextBuilder().withInputStreamWrapper(new FSDataInputStreamWrapper(is)) + .withReaderType(ReaderContext.ReaderType.PREAD).withFileSize(totalDataBlockBytes) + .withFilePath(path).withFileSystem(fs).build(); + HFileBlock.FSReader hbr = + new HFileBlock.FSReaderImpl(context, fileContext, ByteBuffAllocator.HEAP); + + long onDiskSizeOfNextBlock = -1; + long offset = 0; + int numOfReadBlock = 0; + // offset and totalBytes shares the same logic in the HFilePreadReader + while (offset < totalDataBlockBytes) { + HFileBlock block = hbr.readBlockData(offset, onDiskSizeOfNextBlock, true, false, false); + numOfReadBlock++; + try { + onDiskSizeOfNextBlock = block.getNextBlockOnDiskSize(); + offset += block.getOnDiskSizeWithHeader(); + } finally { + block.release(); + } + } + assertEquals(totalDataBlockBytes, offset); + assertEquals(NUM_TEST_BLOCKS, numOfReadBlock); + deleteFile(fs, path); + } + + private void deleteFile(FileSystem fs, Path path) throws IOException { + if (fs.exists(path)) { + fs.delete(path, true); + } + } + @Test public void testReadWithExtra() throws IOException { FileSystem fs = TEST_UTIL.getTestFileSystem();