Source code

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.assertFalse;
022import static org.junit.Assert.assertTrue;
023
024import java.io.IOException;
025import java.util.ArrayList;
026import java.util.List;
027import org.apache.hadoop.conf.Configuration;
028import org.apache.hadoop.fs.Path;
029import org.apache.hadoop.hbase.ByteBufferKeyValue;
030import org.apache.hadoop.hbase.Cell;
031import org.apache.hadoop.hbase.HBaseClassTestRule;
032import org.apache.hadoop.hbase.HBaseTestingUtility;
033import org.apache.hadoop.hbase.HConstants;
034import org.apache.hadoop.hbase.HRegionInfo;
035import org.apache.hadoop.hbase.KeyValue;
036import org.apache.hadoop.hbase.PrivateCellUtil;
037import org.apache.hadoop.hbase.TableName;
038import org.apache.hadoop.hbase.client.Durability;
039import org.apache.hadoop.hbase.client.Put;
040import org.apache.hadoop.hbase.client.Scan;
041import org.apache.hadoop.hbase.regionserver.ChunkCreator;
042import org.apache.hadoop.hbase.regionserver.HRegion;
043import org.apache.hadoop.hbase.regionserver.HStore;
044import org.apache.hadoop.hbase.regionserver.InternalScanner;
045import org.apache.hadoop.hbase.regionserver.MemStoreLABImpl;
046import org.apache.hadoop.hbase.testclassification.MediumTests;
047import org.apache.hadoop.hbase.testclassification.RegionServerTests;
048import org.apache.hadoop.hbase.util.Bytes;
049import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
050import org.apache.hadoop.hbase.wal.WAL;
051import org.junit.After;
052import org.junit.ClassRule;
053import org.junit.Rule;
054import org.junit.Test;
055import org.junit.experimental.categories.Category;
056import org.junit.rules.TestName;
057import org.slf4j.Logger;
058import org.slf4j.LoggerFactory;
059
060@Category({ RegionServerTests.class, MediumTests.class })
061public class TestScannerFromBucketCache {
062
063  @ClassRule
064  public static final HBaseClassTestRule CLASS_RULE =
065      HBaseClassTestRule.forClass(TestScannerFromBucketCache.class);
066
067  private static final Logger LOG = LoggerFactory.getLogger(TestScannerFromBucketCache.class);
068  @Rule
069  public TestName name = new TestName();
070
071  HRegion region = null;
072  private HBaseTestingUtility test_util;
073  public Configuration conf;
074  private final int MAX_VERSIONS = 2;
075  byte[] val = new byte[512 * 1024];
076
077  // Test names
078  private TableName tableName;
079
080  private void setUp(boolean useBucketCache) throws IOException {
081    test_util = HBaseTestingUtility.createLocalHTU();
082    conf = test_util.getConfiguration();
083    if (useBucketCache) {
084      conf.setInt("hbase.bucketcache.size", 400);
085      conf.setStrings(HConstants.BUCKET_CACHE_IOENGINE_KEY, "offheap");
086      conf.setInt("hbase.bucketcache.writer.threads", 10);
087      conf.setFloat("hfile.block.cache.size", 0.2f);
088      conf.setFloat("hbase.regionserver.global.memstore.size", 0.1f);
089    }
090    tableName = TableName.valueOf(name.getMethodName());
091  }
092
093  @After
094  public void tearDown() throws Exception {
095    EnvironmentEdgeManagerTestHelper.reset();
096    LOG.info("Cleaning test directory: " + test_util.getDataTestDir());
097    test_util.cleanupTestDir();
098    CacheConfig.clearGlobalInstances();
099  }
100
101  String getName() {
102    return name.getMethodName();
103  }
104
105  @Test
106  public void testBasicScanWithLRUCache() throws IOException {
107    setUp(false);
108    byte[] row1 = Bytes.toBytes("row1");
109    byte[] qf1 = Bytes.toBytes("qualifier1");
110    byte[] qf2 = Bytes.toBytes("qualifier2");
111    byte[] fam1 = Bytes.toBytes("lrucache");
112
113    long ts1 = 1; // System.currentTimeMillis();
114    long ts2 = ts1 + 1;
115    long ts3 = ts1 + 2;
116
117    // Setting up region
118    String method = this.getName();
119    this.region = initHRegion(tableName, method, conf, test_util, fam1);
120    try {
121      List<Cell> expected = insertData(row1, qf1, qf2, fam1, ts1, ts2, ts3, false);
122
123      List<Cell> actual = performScan(row1, fam1);
124      // Verify result
125      for (int i = 0; i < expected.size(); i++) {
126        assertFalse(actual.get(i) instanceof ByteBufferKeyValue);
127        assertTrue(PrivateCellUtil.equalsIgnoreMvccVersion(expected.get(i), actual.get(i)));
128      }
129      // do the scan again and verify. This time it should be from the lru cache
130      actual = performScan(row1, fam1);
131      // Verify result
132      for (int i = 0; i < expected.size(); i++) {
133        assertFalse(actual.get(i) instanceof ByteBufferKeyValue);
134        assertTrue(PrivateCellUtil.equalsIgnoreMvccVersion(expected.get(i), actual.get(i)));
135      }
136
137    } finally {
138      HBaseTestingUtility.closeRegionAndWAL(this.region);
139      this.region = null;
140    }
141  }
142
143  @Test
144  public void testBasicScanWithOffheapBucketCache() throws IOException {
145    setUp(true);
146    byte[] row1 = Bytes.toBytes("row1offheap");
147    byte[] qf1 = Bytes.toBytes("qualifier1");
148    byte[] qf2 = Bytes.toBytes("qualifier2");
149    byte[] fam1 = Bytes.toBytes("famoffheap");
150
151    long ts1 = 1; // System.currentTimeMillis();
152    long ts2 = ts1 + 1;
153    long ts3 = ts1 + 2;
154
155    // Setting up region
156    String method = this.getName();
157    this.region = initHRegion(tableName, method, conf, test_util, fam1);
158    try {
159      List<Cell> expected = insertData(row1, qf1, qf2, fam1, ts1, ts2, ts3, false);
160
161      List<Cell> actual = performScan(row1, fam1);
162      // Verify result
163      for (int i = 0; i < expected.size(); i++) {
164        assertFalse(actual.get(i) instanceof ByteBufferKeyValue);
165        assertTrue(PrivateCellUtil.equalsIgnoreMvccVersion(expected.get(i), actual.get(i)));
166      }
167      // Wait for the bucket cache threads to move the data to offheap
168      Thread.sleep(500);
169      // do the scan again and verify. This time it should be from the bucket cache in offheap mode
170      actual = performScan(row1, fam1);
171      // Verify result
172      for (int i = 0; i < expected.size(); i++) {
173        assertTrue(actual.get(i) instanceof ByteBufferKeyValue);
174        assertTrue(PrivateCellUtil.equalsIgnoreMvccVersion(expected.get(i), actual.get(i)));
175      }
176
177    } catch (InterruptedException e) {
178    } finally {
179      HBaseTestingUtility.closeRegionAndWAL(this.region);
180      this.region = null;
181    }
182  }
183
184  @Test
185  public void testBasicScanWithOffheapBucketCacheWithMBB() throws IOException {
186    setUp(true);
187    byte[] row1 = Bytes.toBytes("row1offheap");
188    byte[] qf1 = Bytes.toBytes("qualifier1");
189    byte[] qf2 = Bytes.toBytes("qualifier2");
190    byte[] fam1 = Bytes.toBytes("famoffheap");
191
192    long ts1 = 1; // System.currentTimeMillis();
193    long ts2 = ts1 + 1;
194    long ts3 = ts1 + 2;
195
196    // Setting up region
197    String method = this.getName();
198    this.region = initHRegion(tableName, method, conf, test_util, fam1);
199    try {
200      List<Cell> expected = insertData(row1, qf1, qf2, fam1, ts1, ts2, ts3, true);
201
202      List<Cell> actual = performScan(row1, fam1);
203      // Verify result
204      for (int i = 0; i < expected.size(); i++) {
205        assertFalse(actual.get(i) instanceof ByteBufferKeyValue);
206        assertTrue(PrivateCellUtil.equalsIgnoreMvccVersion(expected.get(i), actual.get(i)));
207      }
208      // Wait for the bucket cache threads to move the data to offheap
209      Thread.sleep(500);
210      // do the scan again and verify. This time it should be from the bucket cache in offheap mode
211      // but one of the cell will be copied due to the asSubByteBuff call
212      Scan scan = new Scan(row1);
213      scan.addFamily(fam1);
214      scan.setMaxVersions(10);
215      actual = new ArrayList<>();
216      InternalScanner scanner = region.getScanner(scan);
217
218      boolean hasNext = scanner.next(actual);
219      assertEquals(false, hasNext);
220      // Verify result
221      for (int i = 0; i < expected.size(); i++) {
222        if (i != 5) {
223          // the last cell fetched will be of type shareable but not offheap because
224          // the MBB is copied to form a single cell
225          assertTrue(actual.get(i) instanceof ByteBufferKeyValue);
226        }
227      }
228
229    } catch (InterruptedException e) {
230    } finally {
231      HBaseTestingUtility.closeRegionAndWAL(this.region);
232      this.region = null;
233    }
234  }
235
236  private List<Cell> insertData(byte[] row1, byte[] qf1, byte[] qf2, byte[] fam1, long ts1,
237      long ts2, long ts3, boolean withVal) throws IOException {
238    // Putting data in Region
239    Put put = null;
240    KeyValue kv13 = null;
241    KeyValue kv12 = null;
242    KeyValue kv11 = null;
243
244    KeyValue kv23 = null;
245    KeyValue kv22 = null;
246    KeyValue kv21 = null;
247    if (!withVal) {
248      kv13 = new KeyValue(row1, fam1, qf1, ts3, KeyValue.Type.Put, null);
249      kv12 = new KeyValue(row1, fam1, qf1, ts2, KeyValue.Type.Put, null);
250      kv11 = new KeyValue(row1, fam1, qf1, ts1, KeyValue.Type.Put, null);
251
252      kv23 = new KeyValue(row1, fam1, qf2, ts3, KeyValue.Type.Put, null);
253      kv22 = new KeyValue(row1, fam1, qf2, ts2, KeyValue.Type.Put, null);
254      kv21 = new KeyValue(row1, fam1, qf2, ts1, KeyValue.Type.Put, null);
255    } else {
256      kv13 = new KeyValue(row1, fam1, qf1, ts3, KeyValue.Type.Put, val);
257      kv12 = new KeyValue(row1, fam1, qf1, ts2, KeyValue.Type.Put, val);
258      kv11 = new KeyValue(row1, fam1, qf1, ts1, KeyValue.Type.Put, val);
259
260      kv23 = new KeyValue(row1, fam1, qf2, ts3, KeyValue.Type.Put, val);
261      kv22 = new KeyValue(row1, fam1, qf2, ts2, KeyValue.Type.Put, val);
262      kv21 = new KeyValue(row1, fam1, qf2, ts1, KeyValue.Type.Put, val);
263    }
264
265    put = new Put(row1);
266    put.add(kv13);
267    put.add(kv12);
268    put.add(kv11);
269    put.add(kv23);
270    put.add(kv22);
271    put.add(kv21);
272    region.put(put);
273    region.flush(true);
274    HStore store = region.getStore(fam1);
275    while (store.getStorefilesCount() <= 0) {
276      try {
277        Thread.sleep(20);
278      } catch (InterruptedException e) {
279      }
280    }
281
282    // Expected
283    List<Cell> expected = new ArrayList<>();
284    expected.add(kv13);
285    expected.add(kv12);
286    expected.add(kv23);
287    expected.add(kv22);
288    return expected;
289  }
290
291  private List<Cell> performScan(byte[] row1, byte[] fam1) throws IOException {
292    Scan scan = new Scan(row1);
293    scan.addFamily(fam1);
294    scan.setMaxVersions(MAX_VERSIONS);
295    List<Cell> actual = new ArrayList<>();
296    InternalScanner scanner = region.getScanner(scan);
297
298    boolean hasNext = scanner.next(actual);
299    assertEquals(false, hasNext);
300    return actual;
301  }
302
303  private static HRegion initHRegion(TableName tableName, String callingMethod, Configuration conf,
304      HBaseTestingUtility test_util, byte[]... families) throws IOException {
305    return initHRegion(tableName, null, null, callingMethod, conf, test_util, false, families);
306  }
307
308  private static HRegion initHRegion(TableName tableName, byte[] startKey, byte[] stopKey,
309      String callingMethod, Configuration conf, HBaseTestingUtility test_util, boolean isReadOnly,
310      byte[]... families) throws IOException {
311    ChunkCreator.initialize(MemStoreLABImpl.CHUNK_SIZE_DEFAULT, false, 0, 0, 0, null);
312    Path logDir = test_util.getDataTestDirOnTestFS(callingMethod + ".log");
313    HRegionInfo hri = new HRegionInfo(tableName, startKey, stopKey);
314    final WAL wal = HBaseTestingUtility.createWal(conf, logDir, hri);
315    return initHRegion(tableName, startKey, stopKey, callingMethod, conf, test_util, isReadOnly,
316      Durability.SYNC_WAL, wal, families);
317  }
318
319  /**
320   * @param tableName
321   * @param startKey
322   * @param stopKey
323   * @param callingMethod
324   * @param conf
325   * @param isReadOnly
326   * @param families
327   * @throws IOException
328   * @return A region on which you must call {@link HBaseTestingUtility#closeRegionAndWAL(HRegion)}
329   *         when done.
330   */
331  private static HRegion initHRegion(TableName tableName, byte[] startKey, byte[] stopKey,
332      String callingMethod, Configuration conf, HBaseTestingUtility test_util, boolean isReadOnly,
333      Durability durability, WAL wal, byte[]... families) throws IOException {
334    return test_util.createLocalHRegion(tableName, startKey, stopKey, isReadOnly, durability, wal,
335      families);
336  }
337}