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.assertTrue; 022 023import java.io.IOException; 024import java.util.concurrent.atomic.AtomicInteger; 025import org.apache.hadoop.conf.Configuration; 026import org.apache.hadoop.fs.FSDataOutputStream; 027import org.apache.hadoop.fs.FileSystem; 028import org.apache.hadoop.fs.Path; 029import org.apache.hadoop.hbase.Cell; 030import org.apache.hadoop.hbase.HBaseClassTestRule; 031import org.apache.hadoop.hbase.HBaseTestingUtil; 032import org.apache.hadoop.hbase.KeyValue; 033import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache; 034import org.apache.hadoop.hbase.testclassification.IOTests; 035import org.apache.hadoop.hbase.testclassification.SmallTests; 036import org.apache.hadoop.hbase.util.Bytes; 037import org.junit.ClassRule; 038import org.junit.Test; 039import org.junit.experimental.categories.Category; 040 041import org.apache.hbase.thirdparty.com.google.common.collect.Lists; 042 043/** 044 * Test 045 */ 046@Category({ IOTests.class, SmallTests.class }) 047public class TestHFileReaderImpl { 048 049 @ClassRule 050 public static final HBaseClassTestRule CLASS_RULE = 051 HBaseClassTestRule.forClass(TestHFileReaderImpl.class); 052 053 private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 054 055 static KeyValue toKV(String row) { 056 return new KeyValue(Bytes.toBytes(row), Bytes.toBytes("family"), Bytes.toBytes("qualifier"), 057 Bytes.toBytes("value")); 058 } 059 060 static String toRowStr(Cell c) { 061 return Bytes.toString(c.getRowArray(), c.getRowOffset(), c.getRowLength()); 062 } 063 064 Path makeNewFile() throws IOException { 065 Path ncTFile = new Path(TEST_UTIL.getDataTestDir(), "basic.hfile"); 066 FSDataOutputStream fout = TEST_UTIL.getTestFileSystem().create(ncTFile); 067 int blocksize = toKV("a").getLength() * 3; 068 HFileContext context = 069 new HFileContextBuilder().withBlockSize(blocksize).withIncludesTags(true).build(); 070 Configuration conf = TEST_UTIL.getConfiguration(); 071 HFile.Writer writer = 072 HFile.getWriterFactoryNoCache(conf).withOutputStream(fout).withFileContext(context).create(); 073 // 4 bytes * 3 * 2 for each key/value + 074 // 3 for keys, 15 for values = 42 (woot) 075 writer.append(toKV("c")); 076 writer.append(toKV("e")); 077 writer.append(toKV("g")); 078 // block transition 079 writer.append(toKV("i")); 080 writer.append(toKV("k")); 081 writer.close(); 082 fout.close(); 083 return ncTFile; 084 } 085 086 /** 087 * Test that we only count block size once per block while scanning 088 */ 089 @Test 090 public void testRecordBlockSize() throws IOException { 091 Path p = makeNewFile(); 092 FileSystem fs = TEST_UTIL.getTestFileSystem(); 093 Configuration conf = TEST_UTIL.getConfiguration(); 094 HFile.Reader reader = HFile.createReader(fs, p, CacheConfig.DISABLED, true, conf); 095 096 try (HFileReaderImpl.HFileScannerImpl scanner = 097 (HFileReaderImpl.HFileScannerImpl) reader.getScanner(conf, true, true, false)) { 098 scanner.seekTo(); 099 100 scanner.recordBlockSize( 101 size -> assertTrue("expected non-zero block size on first request", size > 0)); 102 scanner.recordBlockSize( 103 size -> assertEquals("expected zero block size on second request", 0, (int) size)); 104 105 AtomicInteger blocks = new AtomicInteger(0); 106 while (scanner.next()) { 107 scanner.recordBlockSize(size -> { 108 blocks.incrementAndGet(); 109 // there's only 2 cells in the second block 110 assertTrue("expected remaining block to be less than block size", 111 size < toKV("a").getLength() * 3); 112 }); 113 } 114 115 assertEquals("expected only one remaining block but got " + blocks.get(), 1, blocks.get()); 116 } 117 } 118 119 @Test 120 public void testSeekBefore() throws Exception { 121 Path p = makeNewFile(); 122 FileSystem fs = TEST_UTIL.getTestFileSystem(); 123 Configuration conf = TEST_UTIL.getConfiguration(); 124 int[] bucketSizes = { 512, 2048, 4096, 64 * 1024, 128 * 1024 }; 125 BucketCache bucketcache = 126 new BucketCache("offheap", 128 * 1024 * 1024, 64 * 1024, bucketSizes, 5, 64 * 100, null); 127 128 HFile.Reader reader = HFile.createReader(fs, p, new CacheConfig(conf, bucketcache), true, conf); 129 130 // warm cache 131 HFileScanner scanner = reader.getScanner(conf, true, true); 132 scanner.seekTo(toKV("i")); 133 assertEquals("i", toRowStr(scanner.getCell())); 134 scanner.close(); 135 136 while (bucketcache.getBlockCount() <= 0) { 137 Thread.sleep(10); 138 } 139 140 // reopen again. 141 scanner = reader.getScanner(conf, true, true); 142 scanner.seekTo(toKV("i")); 143 assertEquals("i", toRowStr(scanner.getCell())); 144 scanner.seekBefore(toKV("i")); 145 assertEquals("g", toRowStr(scanner.getCell())); 146 scanner.close(); 147 148 for (CachedBlock cachedBlock : Lists.newArrayList(bucketcache)) { 149 BlockCacheKey cacheKey = 150 new BlockCacheKey(cachedBlock.getFilename(), cachedBlock.getOffset()); 151 int refCount = bucketcache.getRpcRefCount(cacheKey); 152 assertEquals(0, refCount); 153 } 154 155 // case 2 156 scanner = reader.getScanner(conf, true, true); 157 scanner.seekTo(toKV("i")); 158 assertEquals("i", toRowStr(scanner.getCell())); 159 scanner.seekBefore(toKV("c")); 160 scanner.close(); 161 for (CachedBlock cachedBlock : Lists.newArrayList(bucketcache)) { 162 BlockCacheKey cacheKey = 163 new BlockCacheKey(cachedBlock.getFilename(), cachedBlock.getOffset()); 164 int refCount = bucketcache.getRpcRefCount(cacheKey); 165 assertEquals(0, refCount); 166 } 167 168 reader.close(); 169 170 // clear bucketcache 171 for (CachedBlock cachedBlock : Lists.newArrayList(bucketcache)) { 172 BlockCacheKey cacheKey = 173 new BlockCacheKey(cachedBlock.getFilename(), cachedBlock.getOffset()); 174 bucketcache.evictBlock(cacheKey); 175 } 176 bucketcache.shutdown(); 177 178 deleteTestDir(fs); 179 } 180 181 protected void deleteTestDir(FileSystem fs) throws IOException { 182 Path dataTestDir = TEST_UTIL.getDataTestDir(); 183 if (fs.exists(dataTestDir)) { 184 fs.delete(dataTestDir, true); 185 } 186 } 187 188}