001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.io.hfile; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertNotNull; 022import static org.junit.Assert.assertTrue; 023import static org.mockito.ArgumentMatchers.any; 024import static org.mockito.ArgumentMatchers.anyBoolean; 025import static org.mockito.Mockito.mock; 026import static org.mockito.Mockito.when; 027 028import java.io.IOException; 029import java.util.concurrent.atomic.AtomicInteger; 030import org.apache.hadoop.conf.Configuration; 031import org.apache.hadoop.fs.FSDataOutputStream; 032import org.apache.hadoop.fs.FileSystem; 033import org.apache.hadoop.fs.Path; 034import org.apache.hadoop.hbase.Cell; 035import org.apache.hadoop.hbase.HBaseClassTestRule; 036import org.apache.hadoop.hbase.HBaseTestingUtil; 037import org.apache.hadoop.hbase.KeyValue; 038import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache; 039import org.apache.hadoop.hbase.testclassification.IOTests; 040import org.apache.hadoop.hbase.testclassification.SmallTests; 041import org.apache.hadoop.hbase.util.Bytes; 042import org.junit.ClassRule; 043import org.junit.Test; 044import org.junit.experimental.categories.Category; 045 046import org.apache.hbase.thirdparty.com.google.common.collect.Lists; 047 048/** 049 * Test 050 */ 051@Category({ IOTests.class, SmallTests.class }) 052public class TestHFileReaderImpl { 053 054 @ClassRule 055 public static final HBaseClassTestRule CLASS_RULE = 056 HBaseClassTestRule.forClass(TestHFileReaderImpl.class); 057 058 private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 059 060 static KeyValue toKV(String row) { 061 return new KeyValue(Bytes.toBytes(row), Bytes.toBytes("family"), Bytes.toBytes("qualifier"), 062 Bytes.toBytes("value")); 063 } 064 065 static String toRowStr(Cell c) { 066 return Bytes.toString(c.getRowArray(), c.getRowOffset(), c.getRowLength()); 067 } 068 069 Path makeNewFile() throws IOException { 070 Path ncTFile = new Path(TEST_UTIL.getDataTestDir(), "basic.hfile"); 071 FSDataOutputStream fout = TEST_UTIL.getTestFileSystem().create(ncTFile); 072 int blocksize = toKV("a").getLength() * 3; 073 HFileContext context = 074 new HFileContextBuilder().withBlockSize(blocksize).withIncludesTags(true).build(); 075 Configuration conf = TEST_UTIL.getConfiguration(); 076 HFile.Writer writer = 077 HFile.getWriterFactoryNoCache(conf).withOutputStream(fout).withFileContext(context).create(); 078 // 4 bytes * 3 * 2 for each key/value + 079 // 3 for keys, 15 for values = 42 (woot) 080 writer.append(toKV("c")); 081 writer.append(toKV("e")); 082 writer.append(toKV("g")); 083 // block transition 084 writer.append(toKV("i")); 085 writer.append(toKV("k")); 086 writer.close(); 087 fout.close(); 088 return ncTFile; 089 } 090 091 /** 092 * Test that we only count block size once per block while scanning 093 */ 094 @Test 095 public void testRecordBlockSize() throws IOException { 096 Path p = makeNewFile(); 097 FileSystem fs = TEST_UTIL.getTestFileSystem(); 098 Configuration conf = TEST_UTIL.getConfiguration(); 099 HFile.Reader reader = HFile.createReader(fs, p, CacheConfig.DISABLED, true, conf); 100 101 try (HFileReaderImpl.HFileScannerImpl scanner = 102 (HFileReaderImpl.HFileScannerImpl) reader.getScanner(conf, true, true, false)) { 103 scanner.seekTo(); 104 105 scanner.recordBlockSize( 106 size -> assertTrue("expected non-zero block size on first request", size > 0)); 107 scanner.recordBlockSize( 108 size -> assertEquals("expected zero block size on second request", 0, (int) size)); 109 110 AtomicInteger blocks = new AtomicInteger(0); 111 while (scanner.next()) { 112 scanner.recordBlockSize(size -> { 113 blocks.incrementAndGet(); 114 // there's only 2 cells in the second block 115 assertTrue("expected remaining block to be less than block size", 116 size < toKV("a").getLength() * 3); 117 }); 118 } 119 120 assertEquals("expected only one remaining block but got " + blocks.get(), 1, blocks.get()); 121 } 122 } 123 124 @Test 125 public void testReadWorksWhenCacheCorrupt() throws Exception { 126 BlockCache mockedCache = mock(BlockCache.class); 127 when(mockedCache.getBlock(any(), anyBoolean(), anyBoolean(), anyBoolean(), any())) 128 .thenThrow(new RuntimeException("Injected error")); 129 Path p = makeNewFile(); 130 FileSystem fs = TEST_UTIL.getTestFileSystem(); 131 Configuration conf = TEST_UTIL.getConfiguration(); 132 HFile.Reader reader = HFile.createReader(fs, p, new CacheConfig(conf, mockedCache), true, conf); 133 long offset = 0; 134 while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) { 135 HFileBlock block = reader.readBlock(offset, -1, false, true, false, true, null, null, false); 136 assertNotNull(block); 137 offset += block.getOnDiskSizeWithHeader(); 138 } 139 } 140 141 @Test 142 public void testSeekBefore() throws Exception { 143 Path p = makeNewFile(); 144 FileSystem fs = TEST_UTIL.getTestFileSystem(); 145 Configuration conf = TEST_UTIL.getConfiguration(); 146 int[] bucketSizes = { 512, 2048, 4096, 64 * 1024, 128 * 1024 }; 147 BucketCache bucketcache = 148 new BucketCache("offheap", 128 * 1024 * 1024, 64 * 1024, bucketSizes, 5, 64 * 100, null); 149 150 HFile.Reader reader = HFile.createReader(fs, p, new CacheConfig(conf, bucketcache), true, conf); 151 152 // warm cache 153 HFileScanner scanner = reader.getScanner(conf, true, true); 154 scanner.seekTo(toKV("i")); 155 assertEquals("i", toRowStr(scanner.getCell())); 156 scanner.close(); 157 158 while (bucketcache.getBlockCount() <= 0) { 159 Thread.sleep(10); 160 } 161 162 // reopen again. 163 scanner = reader.getScanner(conf, true, true); 164 scanner.seekTo(toKV("i")); 165 assertEquals("i", toRowStr(scanner.getCell())); 166 scanner.seekBefore(toKV("i")); 167 assertEquals("g", toRowStr(scanner.getCell())); 168 scanner.close(); 169 170 for (CachedBlock cachedBlock : Lists.newArrayList(bucketcache)) { 171 BlockCacheKey cacheKey = 172 new BlockCacheKey(cachedBlock.getFilename(), cachedBlock.getOffset()); 173 int refCount = bucketcache.getRpcRefCount(cacheKey); 174 assertEquals(0, refCount); 175 } 176 177 // case 2 178 scanner = reader.getScanner(conf, true, true); 179 scanner.seekTo(toKV("i")); 180 assertEquals("i", toRowStr(scanner.getCell())); 181 scanner.seekBefore(toKV("c")); 182 scanner.close(); 183 for (CachedBlock cachedBlock : Lists.newArrayList(bucketcache)) { 184 BlockCacheKey cacheKey = 185 new BlockCacheKey(cachedBlock.getFilename(), cachedBlock.getOffset()); 186 int refCount = bucketcache.getRpcRefCount(cacheKey); 187 assertEquals(0, refCount); 188 } 189 190 reader.close(); 191 192 // clear bucketcache 193 for (CachedBlock cachedBlock : Lists.newArrayList(bucketcache)) { 194 BlockCacheKey cacheKey = 195 new BlockCacheKey(cachedBlock.getFilename(), cachedBlock.getOffset()); 196 bucketcache.evictBlock(cacheKey); 197 } 198 bucketcache.shutdown(); 199 200 deleteTestDir(fs); 201 } 202 203 protected void deleteTestDir(FileSystem fs) throws IOException { 204 Path dataTestDir = TEST_UTIL.getDataTestDir(); 205 if (fs.exists(dataTestDir)) { 206 fs.delete(dataTestDir, true); 207 } 208 } 209 210}