1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.regionserver;
21
22 import static org.junit.Assert.assertEquals;
23 import static org.junit.Assert.assertNull;
24 import static org.junit.Assert.assertTrue;
25 import static org.mockito.Matchers.any;
26 import static org.mockito.Mockito.spy;
27 import static org.mockito.Mockito.times;
28 import static org.mockito.Mockito.verify;
29
30 import com.google.common.collect.Lists;
31 import java.io.IOException;
32 import java.lang.ref.SoftReference;
33 import java.security.PrivilegedExceptionAction;
34 import java.util.ArrayList;
35 import java.util.Arrays;
36 import java.util.Collection;
37 import java.util.Collections;
38 import java.util.Iterator;
39 import java.util.List;
40 import java.util.ListIterator;
41 import java.util.NavigableSet;
42 import java.util.TreeSet;
43 import java.util.concurrent.ConcurrentSkipListSet;
44 import java.util.concurrent.ExecutorService;
45 import java.util.concurrent.Executors;
46 import java.util.concurrent.TimeUnit;
47 import java.util.concurrent.atomic.AtomicBoolean;
48 import org.apache.commons.logging.Log;
49 import org.apache.commons.logging.LogFactory;
50 import org.apache.hadoop.conf.Configuration;
51 import org.apache.hadoop.fs.FSDataOutputStream;
52 import org.apache.hadoop.fs.FileStatus;
53 import org.apache.hadoop.fs.FileSystem;
54 import org.apache.hadoop.fs.FilterFileSystem;
55 import org.apache.hadoop.fs.LocalFileSystem;
56 import org.apache.hadoop.fs.Path;
57 import org.apache.hadoop.fs.permission.FsPermission;
58 import org.apache.hadoop.hbase.Cell;
59 import org.apache.hadoop.hbase.CellUtil;
60 import org.apache.hadoop.hbase.HBaseConfiguration;
61 import org.apache.hadoop.hbase.HBaseTestingUtility;
62 import org.apache.hadoop.hbase.HColumnDescriptor;
63 import org.apache.hadoop.hbase.HRegionInfo;
64 import org.apache.hadoop.hbase.HTableDescriptor;
65 import org.apache.hadoop.hbase.KeyValue;
66 import org.apache.hadoop.hbase.KeyValue.KVComparator;
67 import org.apache.hadoop.hbase.KeyValueUtil;
68 import org.apache.hadoop.hbase.TableName;
69 import org.apache.hadoop.hbase.client.Get;
70 import org.apache.hadoop.hbase.client.Scan;
71 import org.apache.hadoop.hbase.filter.Filter;
72 import org.apache.hadoop.hbase.filter.FilterBase;
73 import org.apache.hadoop.hbase.io.compress.Compression;
74 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
75 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
76 import org.apache.hadoop.hbase.io.hfile.HFile;
77 import org.apache.hadoop.hbase.io.hfile.HFileContext;
78 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
79 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
80 import static org.apache.hadoop.hbase.regionserver.MemStoreChunkPool.CHUNK_POOL_MAXSIZE_KEY;
81 import org.apache.hadoop.hbase.regionserver.compactions.CompactionConfiguration;
82 import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
83 import org.apache.hadoop.hbase.regionserver.querymatcher.ScanQueryMatcher;
84 import org.apache.hadoop.hbase.regionserver.throttle.NoLimitThroughputController;
85 import org.apache.hadoop.hbase.security.User;
86 import org.apache.hadoop.hbase.testclassification.MediumTests;
87 import org.apache.hadoop.hbase.util.Bytes;
88 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
89 import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
90 import org.apache.hadoop.hbase.util.FSUtils;
91 import org.apache.hadoop.hbase.util.IncrementingEnvironmentEdge;
92 import org.apache.hadoop.hbase.util.ManualEnvironmentEdge;
93 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
94 import org.apache.hadoop.hbase.wal.WALFactory;
95 import org.apache.hadoop.util.Progressable;
96 import org.junit.After;
97 import org.junit.Assert;
98 import static org.junit.Assert.assertEquals;
99 import static org.junit.Assert.assertTrue;
100 import org.junit.Before;
101 import org.junit.Rule;
102 import org.junit.Test;
103 import org.junit.experimental.categories.Category;
104 import org.junit.rules.TestName;
105 import org.mockito.Mockito;
106
107
108
109
110 @Category(MediumTests.class)
111 public class TestStore {
112 private static final Log LOG = LogFactory.getLog(TestStore.class);
113 @Rule public TestName name = new TestName();
114
115 HStore store;
116 byte [] table = Bytes.toBytes("table");
117 byte [] family = Bytes.toBytes("family");
118
119 byte [] row = Bytes.toBytes("row");
120 byte [] row2 = Bytes.toBytes("row2");
121 byte [] qf1 = Bytes.toBytes("qf1");
122 byte [] qf2 = Bytes.toBytes("qf2");
123 byte [] qf3 = Bytes.toBytes("qf3");
124 byte [] qf4 = Bytes.toBytes("qf4");
125 byte [] qf5 = Bytes.toBytes("qf5");
126 byte [] qf6 = Bytes.toBytes("qf6");
127
128 NavigableSet<byte[]> qualifiers =
129 new ConcurrentSkipListSet<byte[]>(Bytes.BYTES_COMPARATOR);
130
131 List<Cell> expected = new ArrayList<Cell>();
132 List<Cell> result = new ArrayList<Cell>();
133
134 long id = System.currentTimeMillis();
135 Get get = new Get(row);
136
137 private HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
138 private final String DIR = TEST_UTIL.getDataTestDir("TestStore").toString();
139
140
141
142
143
144
145 @Before
146 public void setUp() throws IOException {
147 qualifiers.add(qf1);
148 qualifiers.add(qf3);
149 qualifiers.add(qf5);
150
151 Iterator<byte[]> iter = qualifiers.iterator();
152 while(iter.hasNext()){
153 byte [] next = iter.next();
154 expected.add(new KeyValue(row, family, next, 1, (byte[])null));
155 get.addColumn(family, next);
156 }
157 }
158
159 private void init(String methodName) throws IOException {
160 init(methodName, TEST_UTIL.getConfiguration());
161 }
162
163 private void init(String methodName, Configuration conf)
164 throws IOException {
165 HColumnDescriptor hcd = new HColumnDescriptor(family);
166
167
168 hcd.setMaxVersions(4);
169 init(methodName, conf, hcd);
170 }
171
172 private void init(String methodName, Configuration conf,
173 HColumnDescriptor hcd) throws IOException {
174 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(table));
175 init(methodName, conf, htd, hcd);
176 }
177
178 private Store init(String methodName, Configuration conf, HTableDescriptor htd,
179 HColumnDescriptor hcd) throws IOException {
180 return init(methodName, conf, htd, hcd, null);
181 }
182
183 @SuppressWarnings("deprecation")
184 private Store init(String methodName, Configuration conf, HTableDescriptor htd,
185 HColumnDescriptor hcd, MyStoreHook hook) throws IOException {
186
187 Path basedir = new Path(DIR+methodName);
188 Path tableDir = FSUtils.getTableDir(basedir, htd.getTableName());
189 final Path logdir = new Path(basedir, DefaultWALProvider.getWALDirectoryName(methodName));
190
191 FileSystem fs = FileSystem.get(conf);
192
193 fs.delete(logdir, true);
194
195 if (htd.hasFamily(hcd.getName())) {
196 htd.modifyFamily(hcd);
197 } else {
198 htd.addFamily(hcd);
199 }
200 HRegionInfo info = new HRegionInfo(htd.getTableName(), null, null, false);
201 final Configuration walConf = new Configuration(conf);
202 FSUtils.setRootDir(walConf, basedir);
203 final WALFactory wals = new WALFactory(walConf, null, methodName);
204 HRegion region = new HRegion(tableDir, wals.getWAL(info.getEncodedNameAsBytes(),
205 info.getTable().getNamespace()), fs, conf, info, htd, null);
206 if (hook == null) {
207 store = new HStore(region, hcd, conf);
208 } else {
209 store = new MyStore(region, hcd, conf, hook);
210 }
211 return store;
212 }
213
214
215
216
217
218
219 @Test
220 public void testFlushSizeAccounting() throws Exception {
221 LOG.info("Setting up a faulty file system that cannot write in " +
222 this.name.getMethodName());
223 final Configuration conf = HBaseConfiguration.create();
224
225 conf.setInt("hbase.hstore.flush.retries.number", 1);
226 User user = User.createUserForTesting(conf, this.name.getMethodName(),
227 new String[]{"foo"});
228
229 conf.setClass("fs.file.impl", FaultyFileSystem.class, FileSystem.class);
230 user.runAs(new PrivilegedExceptionAction<Object>() {
231 @Override
232 public Object run() throws Exception {
233
234 FileSystem fs = FileSystem.get(conf);
235 Assert.assertEquals(FaultyFileSystem.class, fs.getClass());
236 FaultyFileSystem ffs = (FaultyFileSystem)fs;
237
238
239 init(name.getMethodName(), conf);
240
241 long size = store.memstore.getFlushableSize();
242 Assert.assertEquals(0, size);
243 LOG.info("Adding some data");
244 long kvSize = store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
245 size = store.memstore.getFlushableSize();
246 Assert.assertEquals(kvSize, size);
247
248 try {
249 LOG.info("Flushing");
250 flushStore(store, id++);
251 Assert.fail("Didn't bubble up IOE!");
252 } catch (IOException ioe) {
253 Assert.assertTrue(ioe.getMessage().contains("Fault injected"));
254 }
255 size = store.memstore.getFlushableSize();
256 Assert.assertEquals(kvSize, size);
257 store.add(new KeyValue(row, family, qf2, 2, (byte[])null));
258
259
260 Assert.assertEquals(kvSize, size);
261 ffs.fault.set(false);
262 flushStore(store, id++);
263 size = store.memstore.getFlushableSize();
264
265 Assert.assertEquals(kvSize, size);
266 flushStore(store, id++);
267 size = store.memstore.getFlushableSize();
268 Assert.assertEquals(0, size);
269 return null;
270 }
271 });
272 }
273
274
275
276
277
278 @Test
279 public void testCreateWriter() throws Exception {
280 Configuration conf = HBaseConfiguration.create();
281 FileSystem fs = FileSystem.get(conf);
282
283 HColumnDescriptor hcd = new HColumnDescriptor(family);
284 hcd.setCompressionType(Compression.Algorithm.GZ);
285 hcd.setDataBlockEncoding(DataBlockEncoding.DIFF);
286 init(name.getMethodName(), conf, hcd);
287
288
289 StoreFile.Writer writer = store.createWriterInTmp(4, hcd.getCompression(), false, true, false);
290 Path path = writer.getPath();
291 writer.append(new KeyValue(row, family, qf1, Bytes.toBytes(1)));
292 writer.append(new KeyValue(row, family, qf2, Bytes.toBytes(2)));
293 writer.append(new KeyValue(row2, family, qf1, Bytes.toBytes(3)));
294 writer.append(new KeyValue(row2, family, qf2, Bytes.toBytes(4)));
295 writer.close();
296
297
298 HFile.Reader reader = HFile.createReader(fs, path, new CacheConfig(conf), conf);
299 Assert.assertEquals(hcd.getCompressionType(), reader.getCompressionAlgorithm());
300 Assert.assertEquals(hcd.getDataBlockEncoding(), reader.getDataBlockEncoding());
301 reader.close();
302 }
303
304 @Test
305 public void testDeleteExpiredStoreFiles() throws Exception {
306 testDeleteExpiredStoreFiles(0);
307 testDeleteExpiredStoreFiles(1);
308 }
309
310
311
312
313 public void testDeleteExpiredStoreFiles(int minVersions) throws Exception {
314 int storeFileNum = 4;
315 int ttl = 4;
316 IncrementingEnvironmentEdge edge = new IncrementingEnvironmentEdge();
317 EnvironmentEdgeManagerTestHelper.injectEdge(edge);
318
319 Configuration conf = HBaseConfiguration.create();
320
321 conf.setBoolean("hbase.store.delete.expired.storefile", true);
322
323 conf.setInt(CompactionConfiguration.HBASE_HSTORE_COMPACTION_MIN_KEY, 5);
324
325 HColumnDescriptor hcd = new HColumnDescriptor(family);
326 hcd.setMinVersions(minVersions);
327 hcd.setTimeToLive(ttl);
328 init(name.getMethodName() + "-" + minVersions, conf, hcd);
329
330 long storeTtl = this.store.getScanInfo().getTtl();
331 long sleepTime = storeTtl / storeFileNum;
332 long timeStamp;
333
334
335 for (int i = 1; i <= storeFileNum; i++) {
336 LOG.info("Adding some data for the store file #" + i);
337 timeStamp = EnvironmentEdgeManager.currentTime();
338 this.store.add(new KeyValue(row, family, qf1, timeStamp, (byte[]) null));
339 this.store.add(new KeyValue(row, family, qf2, timeStamp, (byte[]) null));
340 this.store.add(new KeyValue(row, family, qf3, timeStamp, (byte[]) null));
341 flush(i);
342 edge.incrementTime(sleepTime);
343 }
344
345
346 Assert.assertEquals(storeFileNum, this.store.getStorefiles().size());
347
348
349
350 for (int i = 1; i <= storeFileNum - 1; i++) {
351
352 assertNull(this.store.requestCompaction());
353 Collection<StoreFile> sfs = this.store.getStorefiles();
354
355 if (minVersions == 0) {
356 assertEquals(storeFileNum - i, sfs.size());
357
358 for (StoreFile sf : sfs) {
359 assertTrue(sf.getReader().getMaxTimestamp() >= (edge.currentTime() - storeTtl));
360 }
361 } else {
362 assertEquals(storeFileNum, sfs.size());
363 }
364
365 edge.incrementTime(sleepTime);
366 }
367 assertNull(this.store.requestCompaction());
368
369 Collection<StoreFile> sfs = this.store.getStorefiles();
370
371 if (minVersions == 0) {
372 assertEquals(1, sfs.size());
373 }
374 long ts = sfs.iterator().next().getReader().getMaxTimestamp();
375 assertTrue(ts < (edge.currentTime() - storeTtl));
376
377 for (StoreFile sf : sfs) {
378 sf.closeReader(true);
379 }
380 }
381
382 @Test
383 public void testRollback() throws IOException {
384 Configuration conf = HBaseConfiguration.create();
385 FileSystem fs = FileSystem.get(conf);
386
387 init(name.getMethodName(), conf);
388 Cell cell = CellUtil.createCell(row, family, qf1);
389 int len = KeyValueUtil.length(cell);
390 int offset = 77;
391 byte[] buf = new byte[offset + len];
392 KeyValueUtil.appendToByteArray(cell, buf, offset);
393 KeyValue newKv = new KeyValue(buf, offset, len);
394 newKv.setSequenceId(cell.getSequenceId());
395 List<Cell> testCells = Arrays.asList(cell, cell, newKv);
396 for (Cell c : testCells) {
397 long sizeBeforeRollback = store.heapSize();
398 store.add(cell);
399 store.rollback(cell);
400 long sizeAeforeRollback = store.heapSize();
401 assertEquals(sizeBeforeRollback, sizeAeforeRollback);
402 }
403 }
404
405 @Test
406 public void testLowestModificationTime() throws Exception {
407 Configuration conf = HBaseConfiguration.create();
408 FileSystem fs = FileSystem.get(conf);
409
410 init(name.getMethodName(), conf);
411
412 int storeFileNum = 4;
413 for (int i = 1; i <= storeFileNum; i++) {
414 LOG.info("Adding some data for the store file #"+i);
415 this.store.add(new KeyValue(row, family, qf1, i, (byte[])null));
416 this.store.add(new KeyValue(row, family, qf2, i, (byte[])null));
417 this.store.add(new KeyValue(row, family, qf3, i, (byte[])null));
418 flush(i);
419 }
420
421 long lowestTimeStampFromManager = StoreUtils.getLowestTimestamp(store.getStorefiles());
422 long lowestTimeStampFromFS = getLowestTimeStampFromFS(fs, store.getStorefiles());
423 Assert.assertEquals(lowestTimeStampFromManager,lowestTimeStampFromFS);
424
425
426 store.compact(store.requestCompaction(), NoLimitThroughputController.INSTANCE);
427 lowestTimeStampFromManager = StoreUtils.getLowestTimestamp(store.getStorefiles());
428 lowestTimeStampFromFS = getLowestTimeStampFromFS(fs, store.getStorefiles());
429 Assert.assertEquals(lowestTimeStampFromManager, lowestTimeStampFromFS);
430 }
431
432 private static long getLowestTimeStampFromFS(FileSystem fs,
433 final Collection<StoreFile> candidates) throws IOException {
434 long minTs = Long.MAX_VALUE;
435 if (candidates.isEmpty()) {
436 return minTs;
437 }
438 Path[] p = new Path[candidates.size()];
439 int i = 0;
440 for (StoreFile sf : candidates) {
441 p[i] = sf.getPath();
442 ++i;
443 }
444
445 FileStatus[] stats = fs.listStatus(p);
446 if (stats == null || stats.length == 0) {
447 return minTs;
448 }
449 for (FileStatus s : stats) {
450 minTs = Math.min(minTs, s.getModificationTime());
451 }
452 return minTs;
453 }
454
455
456
457
458
459 private static final int BLOCKSIZE_SMALL = 8192;
460
461
462
463
464 @Test
465 public void testEmptyStoreFile() throws IOException {
466 init(this.name.getMethodName());
467
468 this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
469 this.store.add(new KeyValue(row, family, qf2, 1, (byte[])null));
470 flush(1);
471
472
473 StoreFile f = this.store.getStorefiles().iterator().next();
474 Path storedir = f.getPath().getParent();
475 long seqid = f.getMaxSequenceId();
476 Configuration c = HBaseConfiguration.create();
477 FileSystem fs = FileSystem.get(c);
478 HFileContext meta = new HFileContextBuilder().withBlockSize(BLOCKSIZE_SMALL).build();
479 StoreFile.Writer w = new StoreFile.WriterBuilder(c, new CacheConfig(c),
480 fs)
481 .withOutputDir(storedir)
482 .withFileContext(meta)
483 .build();
484 w.appendMetadata(seqid + 1, false);
485 w.close();
486 this.store.close();
487
488 this.store = new HStore(this.store.getHRegion(), this.store.getFamily(), c);
489 Assert.assertEquals(2, this.store.getStorefilesCount());
490
491 result = HBaseTestingUtility.getFromStoreFile(store,
492 get.getRow(),
493 qualifiers);
494 Assert.assertEquals(1, result.size());
495 }
496
497
498
499
500
501 @Test
502 public void testGet_FromMemStoreOnly() throws IOException {
503 init(this.name.getMethodName());
504
505
506 this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
507 this.store.add(new KeyValue(row, family, qf2, 1, (byte[])null));
508 this.store.add(new KeyValue(row, family, qf3, 1, (byte[])null));
509 this.store.add(new KeyValue(row, family, qf4, 1, (byte[])null));
510 this.store.add(new KeyValue(row, family, qf5, 1, (byte[])null));
511 this.store.add(new KeyValue(row, family, qf6, 1, (byte[])null));
512
513
514 result = HBaseTestingUtility.getFromStoreFile(store,
515 get.getRow(), qualifiers);
516
517
518 assertCheck();
519 }
520
521
522
523
524
525 @Test
526 public void testGet_FromFilesOnly() throws IOException {
527 init(this.name.getMethodName());
528
529
530 this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
531 this.store.add(new KeyValue(row, family, qf2, 1, (byte[])null));
532
533 flush(1);
534
535
536 this.store.add(new KeyValue(row, family, qf3, 1, (byte[])null));
537 this.store.add(new KeyValue(row, family, qf4, 1, (byte[])null));
538
539 flush(2);
540
541
542 this.store.add(new KeyValue(row, family, qf5, 1, (byte[])null));
543 this.store.add(new KeyValue(row, family, qf6, 1, (byte[])null));
544
545 flush(3);
546
547
548 result = HBaseTestingUtility.getFromStoreFile(store,
549 get.getRow(),
550 qualifiers);
551
552
553
554 Collections.sort(result, KeyValue.COMPARATOR);
555
556
557 assertCheck();
558 }
559
560
561
562
563
564 @Test
565 public void testGet_FromMemStoreAndFiles() throws IOException {
566 init(this.name.getMethodName());
567
568
569 this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
570 this.store.add(new KeyValue(row, family, qf2, 1, (byte[])null));
571
572 flush(1);
573
574
575 this.store.add(new KeyValue(row, family, qf3, 1, (byte[])null));
576 this.store.add(new KeyValue(row, family, qf4, 1, (byte[])null));
577
578 flush(2);
579
580
581 this.store.add(new KeyValue(row, family, qf5, 1, (byte[])null));
582 this.store.add(new KeyValue(row, family, qf6, 1, (byte[])null));
583
584
585 result = HBaseTestingUtility.getFromStoreFile(store,
586 get.getRow(), qualifiers);
587
588
589 Collections.sort(result, KeyValue.COMPARATOR);
590
591
592 assertCheck();
593 }
594
595 private void flush(int storeFilessize) throws IOException{
596 this.store.snapshot();
597 flushStore(store, id++);
598 Assert.assertEquals(storeFilessize, this.store.getStorefiles().size());
599 Assert.assertEquals(0, ((DefaultMemStore)this.store.memstore).activeSection.getCellSkipListSet().size());
600 }
601
602 private void assertCheck() {
603 Assert.assertEquals(expected.size(), result.size());
604 for(int i=0; i<expected.size(); i++) {
605 Assert.assertEquals(expected.get(i), result.get(i));
606 }
607 }
608
609
610
611
612
613
614
615 @Test
616 public void testIncrementColumnValue_ICVDuringFlush()
617 throws IOException, InterruptedException {
618 init(this.name.getMethodName());
619
620 long oldValue = 1L;
621 long newValue = 3L;
622 this.store.add(new KeyValue(row, family, qf1,
623 System.currentTimeMillis(),
624 Bytes.toBytes(oldValue)));
625
626
627 this.store.snapshot();
628
629
630 this.store.add(new KeyValue(row, family, qf2,
631 System.currentTimeMillis(),
632 Bytes.toBytes(oldValue)));
633
634
635 long ret = this.store.updateColumnValue(row, family, qf1, newValue);
636
637
638 Assert.assertTrue(ret > 0);
639
640
641 flushStore(store, id++);
642 Assert.assertEquals(1, this.store.getStorefiles().size());
643
644 Assert.assertEquals(2, ((DefaultMemStore)this.store.memstore).activeSection.getCellSkipListSet().size());
645
646
647 Get get = new Get(row);
648 get.addColumn(family, qf1);
649 get.setMaxVersions();
650 List<Cell> results = new ArrayList<Cell>();
651
652 results = HBaseTestingUtility.getFromStoreFile(store, get);
653 Assert.assertEquals(2, results.size());
654
655 long ts1 = results.get(0).getTimestamp();
656 long ts2 = results.get(1).getTimestamp();
657
658 Assert.assertTrue(ts1 > ts2);
659
660 Assert.assertEquals(newValue, Bytes.toLong(CellUtil.cloneValue(results.get(0))));
661 Assert.assertEquals(oldValue, Bytes.toLong(CellUtil.cloneValue(results.get(1))));
662 }
663
664 @After
665 public void tearDown() throws Exception {
666 EnvironmentEdgeManagerTestHelper.reset();
667 }
668
669 @Test
670 public void testICV_negMemstoreSize() throws IOException {
671 init(this.name.getMethodName());
672
673 long time = 100;
674 ManualEnvironmentEdge ee = new ManualEnvironmentEdge();
675 ee.setValue(time);
676 EnvironmentEdgeManagerTestHelper.injectEdge(ee);
677 long newValue = 3L;
678 long size = 0;
679
680
681 size += this.store.add(new KeyValue(Bytes.toBytes("200909091000"), family, qf1,
682 System.currentTimeMillis(), Bytes.toBytes(newValue)));
683 size += this.store.add(new KeyValue(Bytes.toBytes("200909091200"), family, qf1,
684 System.currentTimeMillis(), Bytes.toBytes(newValue)));
685 size += this.store.add(new KeyValue(Bytes.toBytes("200909091300"), family, qf1,
686 System.currentTimeMillis(), Bytes.toBytes(newValue)));
687 size += this.store.add(new KeyValue(Bytes.toBytes("200909091400"), family, qf1,
688 System.currentTimeMillis(), Bytes.toBytes(newValue)));
689 size += this.store.add(new KeyValue(Bytes.toBytes("200909091500"), family, qf1,
690 System.currentTimeMillis(), Bytes.toBytes(newValue)));
691
692
693 for ( int i = 0 ; i < 10000 ; ++i) {
694 newValue++;
695
696 long ret = this.store.updateColumnValue(row, family, qf1, newValue);
697 long ret2 = this.store.updateColumnValue(row2, family, qf1, newValue);
698
699 if (ret != 0) System.out.println("ret: " + ret);
700 if (ret2 != 0) System.out.println("ret2: " + ret2);
701
702 Assert.assertTrue("ret: " + ret, ret >= 0);
703 size += ret;
704 Assert.assertTrue("ret2: " + ret2, ret2 >= 0);
705 size += ret2;
706
707
708 if (i % 1000 == 0)
709 ee.setValue(++time);
710 }
711
712 long computedSize=0;
713 for (Cell cell : ((DefaultMemStore)this.store.memstore).activeSection.getCellSkipListSet()) {
714 long kvsize = DefaultMemStore.heapSizeChange(cell, true);
715
716 computedSize += kvsize;
717 }
718 Assert.assertEquals(computedSize, size);
719 }
720
721 @Test
722 public void testIncrementColumnValue_SnapshotFlushCombo() throws Exception {
723 ManualEnvironmentEdge mee = new ManualEnvironmentEdge();
724 EnvironmentEdgeManagerTestHelper.injectEdge(mee);
725 init(this.name.getMethodName());
726
727 long oldValue = 1L;
728 long newValue = 3L;
729 this.store.add(new KeyValue(row, family, qf1,
730 EnvironmentEdgeManager.currentTime(),
731 Bytes.toBytes(oldValue)));
732
733
734 this.store.snapshot();
735
736
737 long ret = this.store.updateColumnValue(row, family, qf1, newValue);
738
739
740 Assert.assertTrue(ret > 0);
741
742
743 flushStore(store, id++);
744 Assert.assertEquals(1, this.store.getStorefiles().size());
745 Assert.assertEquals(1, ((DefaultMemStore)this.store.memstore).activeSection.getCellSkipListSet().size());
746
747
748 newValue += 1;
749 this.store.updateColumnValue(row, family, qf1, newValue);
750
751
752 newValue += 1;
753 this.store.updateColumnValue(row, family, qf1, newValue);
754
755
756
757
758
759 Get get = new Get(row);
760 get.addColumn(family, qf1);
761 get.setMaxVersions();
762 List<Cell> results = new ArrayList<Cell>();
763
764 results = HBaseTestingUtility.getFromStoreFile(store, get);
765 Assert.assertEquals(2, results.size());
766
767 long ts1 = results.get(0).getTimestamp();
768 long ts2 = results.get(1).getTimestamp();
769
770 Assert.assertTrue(ts1 > ts2);
771 Assert.assertEquals(newValue, Bytes.toLong(CellUtil.cloneValue(results.get(0))));
772 Assert.assertEquals(oldValue, Bytes.toLong(CellUtil.cloneValue(results.get(1))));
773
774 mee.setValue(2);
775 newValue += 1;
776 this.store.updateColumnValue(row, family, qf1, newValue);
777
778 results = HBaseTestingUtility.getFromStoreFile(store, get);
779 Assert.assertEquals(2, results.size());
780
781 ts1 = results.get(0).getTimestamp();
782 ts2 = results.get(1).getTimestamp();
783
784 Assert.assertTrue(ts1 > ts2);
785 Assert.assertEquals(newValue, Bytes.toLong(CellUtil.cloneValue(results.get(0))));
786 Assert.assertEquals(oldValue, Bytes.toLong(CellUtil.cloneValue(results.get(1))));
787 }
788
789 @Test
790 public void testHandleErrorsInFlush() throws Exception {
791 LOG.info("Setting up a faulty file system that cannot write");
792
793 final Configuration conf = HBaseConfiguration.create();
794 User user = User.createUserForTesting(conf,
795 "testhandleerrorsinflush", new String[]{"foo"});
796
797 conf.setClass("fs.file.impl", FaultyFileSystem.class,
798 FileSystem.class);
799 user.runAs(new PrivilegedExceptionAction<Object>() {
800 @Override
801 public Object run() throws Exception {
802
803 FileSystem fs = FileSystem.get(conf);
804 Assert.assertEquals(FaultyFileSystem.class, fs.getClass());
805
806
807 init(name.getMethodName(), conf);
808
809 LOG.info("Adding some data");
810 store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
811 store.add(new KeyValue(row, family, qf2, 1, (byte[])null));
812 store.add(new KeyValue(row, family, qf3, 1, (byte[])null));
813
814 LOG.info("Before flush, we should have no files");
815
816 Collection<StoreFileInfo> files =
817 store.getRegionFileSystem().getStoreFiles(store.getColumnFamilyName());
818 Assert.assertEquals(0, files != null ? files.size() : 0);
819
820
821 try {
822 LOG.info("Flushing");
823 flush(1);
824 Assert.fail("Didn't bubble up IOE!");
825 } catch (IOException ioe) {
826 Assert.assertTrue(ioe.getMessage().contains("Fault injected"));
827 }
828
829 LOG.info("After failed flush, we should still have no files!");
830 files = store.getRegionFileSystem().getStoreFiles(store.getColumnFamilyName());
831 Assert.assertEquals(0, files != null ? files.size() : 0);
832 store.getHRegion().getWAL().close();
833 return null;
834 }
835 });
836 FileSystem.closeAllForUGI(user.getUGI());
837 }
838
839
840
841
842
843 static class FaultyFileSystem extends FilterFileSystem {
844 List<SoftReference<FaultyOutputStream>> outStreams =
845 new ArrayList<SoftReference<FaultyOutputStream>>();
846 private long faultPos = 200;
847 AtomicBoolean fault = new AtomicBoolean(true);
848
849 public FaultyFileSystem() {
850 super(new LocalFileSystem());
851 System.err.println("Creating faulty!");
852 }
853
854 @Override
855 public FSDataOutputStream create(Path p) throws IOException {
856 return new FaultyOutputStream(super.create(p), faultPos, fault);
857 }
858
859 @Override
860 public FSDataOutputStream create(Path f, FsPermission permission,
861 boolean overwrite, int bufferSize, short replication, long blockSize,
862 Progressable progress) throws IOException {
863 return new FaultyOutputStream(super.create(f, permission,
864 overwrite, bufferSize, replication, blockSize, progress), faultPos, fault);
865 }
866
867 @Override
868 public FSDataOutputStream createNonRecursive(Path f, boolean overwrite,
869 int bufferSize, short replication, long blockSize, Progressable progress)
870 throws IOException {
871
872
873 return create(f, overwrite, bufferSize, replication, blockSize, progress);
874 }
875 }
876
877 static class FaultyOutputStream extends FSDataOutputStream {
878 volatile long faultPos = Long.MAX_VALUE;
879 private final AtomicBoolean fault;
880
881 public FaultyOutputStream(FSDataOutputStream out, long faultPos, final AtomicBoolean fault)
882 throws IOException {
883 super(out, null);
884 this.faultPos = faultPos;
885 this.fault = fault;
886 }
887
888 @Override
889 public void write(byte[] buf, int offset, int length) throws IOException {
890 System.err.println("faulty stream write at pos " + getPos());
891 injectFault();
892 super.write(buf, offset, length);
893 }
894
895 private void injectFault() throws IOException {
896 if (this.fault.get() && getPos() >= faultPos) {
897 throw new IOException("Fault injected");
898 }
899 }
900 }
901
902 private static void flushStore(HStore store, long id) throws IOException {
903 StoreFlushContext storeFlushCtx = store.createFlushContext(id);
904 storeFlushCtx.prepare();
905 storeFlushCtx.flushCache(Mockito.mock(MonitoredTask.class));
906 storeFlushCtx.commit(Mockito.mock(MonitoredTask.class));
907 }
908
909
910
911
912
913
914
915
916
917 List<Cell> getKeyValueSet(long[] timestamps, int numRows,
918 byte[] qualifier, byte[] family) {
919 List<Cell> kvList = new ArrayList<Cell>();
920 for (int i=1;i<=numRows;i++) {
921 byte[] b = Bytes.toBytes(i);
922 for (long timestamp: timestamps) {
923 kvList.add(new KeyValue(b, family, qualifier, timestamp, b));
924 }
925 }
926 return kvList;
927 }
928
929
930
931
932
933 @Test
934 public void testMultipleTimestamps() throws IOException {
935 int numRows = 1;
936 long[] timestamps1 = new long[] {1,5,10,20};
937 long[] timestamps2 = new long[] {30,80};
938
939 init(this.name.getMethodName());
940
941 List<Cell> kvList1 = getKeyValueSet(timestamps1,numRows, qf1, family);
942 for (Cell kv : kvList1) {
943 this.store.add(KeyValueUtil.ensureKeyValue(kv));
944 }
945
946 this.store.snapshot();
947 flushStore(store, id++);
948
949 List<Cell> kvList2 = getKeyValueSet(timestamps2,numRows, qf1, family);
950 for(Cell kv : kvList2) {
951 this.store.add(KeyValueUtil.ensureKeyValue(kv));
952 }
953
954 List<Cell> result;
955 Get get = new Get(Bytes.toBytes(1));
956 get.addColumn(family,qf1);
957
958 get.setTimeRange(0,15);
959 result = HBaseTestingUtility.getFromStoreFile(store, get);
960 Assert.assertTrue(result.size()>0);
961
962 get.setTimeRange(40,90);
963 result = HBaseTestingUtility.getFromStoreFile(store, get);
964 Assert.assertTrue(result.size()>0);
965
966 get.setTimeRange(10,45);
967 result = HBaseTestingUtility.getFromStoreFile(store, get);
968 Assert.assertTrue(result.size()>0);
969
970 get.setTimeRange(80,145);
971 result = HBaseTestingUtility.getFromStoreFile(store, get);
972 Assert.assertTrue(result.size()>0);
973
974 get.setTimeRange(1,2);
975 result = HBaseTestingUtility.getFromStoreFile(store, get);
976 Assert.assertTrue(result.size()>0);
977
978 get.setTimeRange(90,200);
979 result = HBaseTestingUtility.getFromStoreFile(store, get);
980 Assert.assertTrue(result.size()==0);
981 }
982
983
984
985
986
987
988 @Test
989 public void testSplitWithEmptyColFam() throws IOException {
990 init(this.name.getMethodName());
991 Assert.assertNull(store.getSplitPoint());
992 store.getHRegion().forceSplit(null);
993 Assert.assertNull(store.getSplitPoint());
994 store.getHRegion().clearSplit();
995 }
996
997 @Test
998 public void testStoreUsesConfigurationFromHcdAndHtd() throws Exception {
999 final String CONFIG_KEY = "hbase.regionserver.thread.compaction.throttle";
1000 long anyValue = 10;
1001
1002
1003
1004
1005 Configuration conf = HBaseConfiguration.create();
1006 conf.setLong(CONFIG_KEY, anyValue);
1007 init(name.getMethodName() + "-xml", conf);
1008 Assert.assertTrue(store.throttleCompaction(anyValue + 1));
1009 Assert.assertFalse(store.throttleCompaction(anyValue));
1010
1011
1012 --anyValue;
1013 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(table));
1014 HColumnDescriptor hcd = new HColumnDescriptor(family);
1015 htd.setConfiguration(CONFIG_KEY, Long.toString(anyValue));
1016 init(name.getMethodName() + "-htd", conf, htd, hcd);
1017 Assert.assertTrue(store.throttleCompaction(anyValue + 1));
1018 Assert.assertFalse(store.throttleCompaction(anyValue));
1019
1020
1021 --anyValue;
1022 hcd.setConfiguration(CONFIG_KEY, Long.toString(anyValue));
1023 init(name.getMethodName() + "-hcd", conf, htd, hcd);
1024 Assert.assertTrue(store.throttleCompaction(anyValue + 1));
1025 Assert.assertFalse(store.throttleCompaction(anyValue));
1026 }
1027
1028 public static class DummyStoreEngine extends DefaultStoreEngine {
1029 public static DefaultCompactor lastCreatedCompactor = null;
1030 @Override
1031 protected void createComponents(
1032 Configuration conf, Store store, KVComparator comparator) throws IOException {
1033 super.createComponents(conf, store, comparator);
1034 lastCreatedCompactor = this.compactor;
1035 }
1036 }
1037
1038 @Test
1039 public void testStoreUsesSearchEngineOverride() throws Exception {
1040 Configuration conf = HBaseConfiguration.create();
1041 conf.set(StoreEngine.STORE_ENGINE_CLASS_KEY, DummyStoreEngine.class.getName());
1042 init(this.name.getMethodName(), conf);
1043 Assert.assertEquals(DummyStoreEngine.lastCreatedCompactor,
1044 this.store.storeEngine.getCompactor());
1045 }
1046
1047 private void addStoreFile() throws IOException {
1048 StoreFile f = this.store.getStorefiles().iterator().next();
1049 Path storedir = f.getPath().getParent();
1050 long seqid = this.store.getMaxSequenceId();
1051 Configuration c = TEST_UTIL.getConfiguration();
1052 FileSystem fs = FileSystem.get(c);
1053 HFileContext fileContext = new HFileContextBuilder().withBlockSize(BLOCKSIZE_SMALL).build();
1054 StoreFile.Writer w = new StoreFile.WriterBuilder(c, new CacheConfig(c),
1055 fs)
1056 .withOutputDir(storedir)
1057 .withFileContext(fileContext)
1058 .build();
1059 w.appendMetadata(seqid + 1, false);
1060 w.close();
1061 LOG.info("Added store file:" + w.getPath());
1062 }
1063
1064 private void archiveStoreFile(int index) throws IOException {
1065 Collection<StoreFile> files = this.store.getStorefiles();
1066 StoreFile sf = null;
1067 Iterator<StoreFile> it = files.iterator();
1068 for (int i = 0; i <= index; i++) {
1069 sf = it.next();
1070 }
1071 store.getRegionFileSystem().removeStoreFiles(store.getColumnFamilyName(), Lists.newArrayList(sf));
1072 }
1073
1074 private void closeCompactedFile(int index) throws IOException {
1075 Collection<StoreFile> files =
1076 this.store.getStoreEngine().getStoreFileManager().getCompactedfiles();
1077 StoreFile sf = null;
1078 Iterator<StoreFile> it = files.iterator();
1079 for (int i = 0; i <= index; i++) {
1080 sf = it.next();
1081 }
1082 sf.closeReader(true);
1083 store.getStoreEngine().getStoreFileManager().removeCompactedFiles(Lists.newArrayList(sf));
1084 }
1085
1086 @Test
1087 public void testRefreshStoreFiles() throws Exception {
1088 init(name.getMethodName());
1089
1090 assertEquals(0, this.store.getStorefilesCount());
1091
1092
1093 store.refreshStoreFiles();
1094 assertEquals(0, this.store.getStorefilesCount());
1095
1096
1097 this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
1098 flush(1);
1099 assertEquals(1, this.store.getStorefilesCount());
1100
1101
1102 addStoreFile();
1103
1104 assertEquals(1, this.store.getStorefilesCount());
1105 store.refreshStoreFiles();
1106 assertEquals(2, this.store.getStorefilesCount());
1107
1108
1109 addStoreFile();
1110 addStoreFile();
1111 addStoreFile();
1112
1113 assertEquals(2, this.store.getStorefilesCount());
1114 store.refreshStoreFiles();
1115 assertEquals(5, this.store.getStorefilesCount());
1116
1117 closeCompactedFile(0);
1118 archiveStoreFile(0);
1119
1120 assertEquals(5, this.store.getStorefilesCount());
1121 store.refreshStoreFiles();
1122 assertEquals(4, this.store.getStorefilesCount());
1123
1124 archiveStoreFile(0);
1125 archiveStoreFile(1);
1126 archiveStoreFile(2);
1127
1128 assertEquals(4, this.store.getStorefilesCount());
1129 store.refreshStoreFiles();
1130 assertEquals(1, this.store.getStorefilesCount());
1131
1132 archiveStoreFile(0);
1133 store.refreshStoreFiles();
1134 assertEquals(0, this.store.getStorefilesCount());
1135 }
1136
1137 @SuppressWarnings("unchecked")
1138 @Test
1139 public void testRefreshStoreFilesNotChanged() throws IOException {
1140 init(name.getMethodName());
1141
1142 assertEquals(0, this.store.getStorefilesCount());
1143
1144
1145 this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
1146 flush(1);
1147
1148 addStoreFile();
1149
1150 HStore spiedStore = spy(store);
1151
1152
1153 spiedStore.refreshStoreFiles();
1154 assertEquals(2, this.store.getStorefilesCount());
1155 verify(spiedStore, times(1)).replaceStoreFiles(any(Collection.class), any(Collection.class));
1156
1157
1158 spiedStore.refreshStoreFiles();
1159
1160
1161 verify(spiedStore, times(0)).replaceStoreFiles(null, null);
1162 }
1163
1164 private Cell createCell(byte[] qualifier, long ts, long sequenceId, byte[] value) throws IOException {
1165 return createCell(row, qualifier, ts, sequenceId, value);
1166 }
1167
1168 private Cell createCell(byte[] row, byte[] qualifier, long ts, long sequenceId, byte[] value) throws IOException {
1169 Cell c = CellUtil.createCell(row, family, qualifier, ts, KeyValue.Type.Put.getCode(), value);
1170 CellUtil.setSequenceId(c, sequenceId);
1171 return c;
1172 }
1173
1174 @Test
1175 public void testScanWithDoubleFlush() throws IOException {
1176 Configuration conf = HBaseConfiguration.create();
1177
1178 MyStore myStore = initMyStore(name.getMethodName(), conf, new MyStoreHook() {
1179 @Override
1180 public void getScanners(final MyStore store) throws IOException {
1181 final long tmpId = id++;
1182 ExecutorService s = Executors.newSingleThreadExecutor();
1183 s.submit(new Runnable() {
1184 @Override
1185 public void run() {
1186 try {
1187
1188
1189
1190
1191 flushStore(store, tmpId);
1192 } catch (IOException ex) {
1193 throw new RuntimeException(ex);
1194 }
1195 }
1196 });
1197 s.shutdown();
1198 try {
1199
1200 s.awaitTermination(500, TimeUnit.MILLISECONDS);
1201 } catch (InterruptedException ex) {
1202 }
1203 }
1204 });
1205 byte[] oldValue = Bytes.toBytes("oldValue");
1206 byte[] currentValue = Bytes.toBytes("currentValue");
1207 long ts = EnvironmentEdgeManager.currentTime();
1208 long seqId = 100;
1209
1210 myStore.add(createCell(qf1, ts, seqId, oldValue));
1211 myStore.add(createCell(qf2, ts, seqId, oldValue));
1212 myStore.add(createCell(qf3, ts, seqId, oldValue));
1213 long snapshotId = id++;
1214
1215 StoreFlushContext storeFlushCtx = store.createFlushContext(snapshotId);
1216 storeFlushCtx.prepare();
1217
1218
1219 myStore.add(createCell(qf1, ts + 1, seqId + 1, currentValue));
1220 myStore.add(createCell(qf2, ts + 1, seqId + 1, currentValue));
1221 myStore.add(createCell(qf3, ts + 1, seqId + 1, currentValue));
1222 TreeSet<byte[]> quals = new TreeSet<>(Bytes.BYTES_COMPARATOR);
1223 quals.add(qf1);
1224 quals.add(qf2);
1225 quals.add(qf3);
1226 try (InternalScanner scanner = (InternalScanner) myStore.getScanner(
1227 new Scan(new Get(row)), quals, seqId + 1)) {
1228
1229 storeFlushCtx.flushCache(Mockito.mock(MonitoredTask.class));
1230 storeFlushCtx.commit(Mockito.mock(MonitoredTask.class));
1231
1232 List<Cell> results = new ArrayList<>();
1233 scanner.next(results);
1234 assertEquals(3, results.size());
1235 for (Cell c : results) {
1236 byte[] actualValue = CellUtil.cloneValue(c);
1237 assertTrue("expected:" + Bytes.toStringBinary(currentValue)
1238 + ", actual:" + Bytes.toStringBinary(actualValue),
1239 Bytes.equals(actualValue, currentValue));
1240 }
1241 }
1242
1243 }
1244
1245 @Test
1246 public void testFlushBeforeCompletingScanWoFilter() throws IOException, InterruptedException {
1247 final AtomicBoolean timeToGoNextRow = new AtomicBoolean(false);
1248 testFlushBeforeCompletingScan(new MyListHook() {
1249 @Override
1250 public void hook(int currentSize) {
1251 if (currentSize == 2) {
1252 try {
1253 flushStore(store, id++);
1254 timeToGoNextRow.set(true);
1255 } catch (IOException e) {
1256 throw new RuntimeException(e);
1257 }
1258 }
1259 }
1260 }, new FilterBase() {
1261 @Override
1262 public Filter.ReturnCode filterKeyValue(Cell v) throws IOException {
1263 return Filter.ReturnCode.INCLUDE;
1264 }
1265 });
1266 }
1267
1268 @Test
1269 public void testFlushBeforeCompletingScanWithFilter() throws IOException, InterruptedException {
1270 final AtomicBoolean timeToGoNextRow = new AtomicBoolean(false);
1271 testFlushBeforeCompletingScan(new MyListHook() {
1272 @Override
1273 public void hook(int currentSize) {
1274 if (currentSize == 2) {
1275 try {
1276 flushStore(store, id++);
1277 timeToGoNextRow.set(true);
1278 } catch (IOException e) {
1279 throw new RuntimeException(e);
1280 }
1281 }
1282 }
1283 }, new FilterBase() {
1284 @Override
1285 public Filter.ReturnCode filterKeyValue(Cell v) throws IOException {
1286 if (timeToGoNextRow.get()) {
1287 timeToGoNextRow.set(false);
1288 return Filter.ReturnCode.NEXT_ROW;
1289 } else {
1290 return Filter.ReturnCode.INCLUDE;
1291 }
1292 }
1293 });
1294 }
1295
1296 @Test
1297 public void testFlushBeforeCompletingScanWithFilterHint() throws IOException, InterruptedException {
1298 final AtomicBoolean timeToGetHint = new AtomicBoolean(false);
1299 testFlushBeforeCompletingScan(new MyListHook() {
1300 @Override
1301 public void hook(int currentSize) {
1302 if (currentSize == 2) {
1303 try {
1304 flushStore(store, id++);
1305 timeToGetHint.set(true);
1306 } catch (IOException e) {
1307 throw new RuntimeException(e);
1308 }
1309 }
1310 }
1311 }, new FilterBase() {
1312 @Override
1313 public Filter.ReturnCode filterKeyValue(Cell v) throws IOException {
1314 if (timeToGetHint.get()) {
1315 timeToGetHint.set(false);
1316 return Filter.ReturnCode.SEEK_NEXT_USING_HINT;
1317 } else {
1318 return Filter.ReturnCode.INCLUDE;
1319 }
1320 }
1321 @Override
1322 public Cell getNextCellHint(Cell currentCell) throws IOException {
1323 return currentCell;
1324 }
1325 });
1326 }
1327
1328 private void testFlushBeforeCompletingScan(MyListHook hook, Filter filter)
1329 throws IOException, InterruptedException {
1330 Configuration conf = HBaseConfiguration.create();
1331 HColumnDescriptor hcd = new HColumnDescriptor(family);
1332 hcd.setMaxVersions(1);
1333 byte[] r0 = Bytes.toBytes("row0");
1334 byte[] r1 = Bytes.toBytes("row1");
1335 byte[] r2 = Bytes.toBytes("row2");
1336 byte[] value0 = Bytes.toBytes("value0");
1337 byte[] value1 = Bytes.toBytes("value1");
1338 byte[] value2 = Bytes.toBytes("value2");
1339 long ts = EnvironmentEdgeManager.currentTime();
1340 final long seqId = 100;
1341 init(name.getMethodName(), conf, new HTableDescriptor(TableName.valueOf(table)), hcd, new MyStoreHook() {
1342 @Override
1343 public long getSmallestReadPoint(HStore store) {
1344 return seqId + 3;
1345 }
1346 });
1347
1348 store.add(createCell(r0, qf1, ts, seqId, value0));
1349 store.add(createCell(r0, qf2, ts, seqId, value0));
1350 store.add(createCell(r0, qf3, ts, seqId, value0));
1351 store.add(createCell(r1, qf1, ts + 1, seqId + 1, value1));
1352 store.add(createCell(r1, qf2, ts + 1, seqId + 1, value1));
1353 store.add(createCell(r1, qf3, ts + 1, seqId + 1, value1));
1354 store.add(createCell(r2, qf1, ts + 2, seqId + 2, value2));
1355 store.add(createCell(r2, qf2, ts + 2, seqId + 2, value2));
1356 store.add(createCell(r2, qf3, ts + 2, seqId + 2, value2));
1357 store.add(createCell(r1, qf1, ts + 3, seqId + 3, value1));
1358 store.add(createCell(r1, qf2, ts + 3, seqId + 3, value1));
1359 store.add(createCell(r1, qf3, ts + 3, seqId + 3, value1));
1360 List<Cell> myList = new MyList<>(hook);
1361 Scan scan = new Scan()
1362 .withStartRow(r1)
1363 .setFilter(filter);
1364 try (InternalScanner scanner = (InternalScanner) store.getScanner(
1365 scan, null, seqId + 3)){
1366
1367 scanner.next(myList);
1368 assertEquals(3, myList.size());
1369 for (Cell c : myList) {
1370 byte[] actualValue = CellUtil.cloneValue(c);
1371 assertTrue("expected:" + Bytes.toStringBinary(value1)
1372 + ", actual:" + Bytes.toStringBinary(actualValue)
1373 , Bytes.equals(actualValue, value1));
1374 }
1375 List<Cell> normalList = new ArrayList<>(3);
1376
1377 scanner.next(normalList);
1378 assertEquals(3, normalList.size());
1379 for (Cell c : normalList) {
1380 byte[] actualValue = CellUtil.cloneValue(c);
1381 assertTrue("expected:" + Bytes.toStringBinary(value2)
1382 + ", actual:" + Bytes.toStringBinary(actualValue)
1383 , Bytes.equals(actualValue, value2));
1384 }
1385 }
1386 }
1387
1388 @Test
1389 public void testReclaimChunkWhenScaning() throws IOException {
1390 Configuration conf = HBaseConfiguration.create();
1391 conf.setFloat(CHUNK_POOL_MAXSIZE_KEY, 1);
1392 init("testReclaimChunkWhenScaning", conf);
1393 final long ts = EnvironmentEdgeManager.currentTime();
1394 final long seqId = 100;
1395 byte[] value = Bytes.toBytes("value");
1396
1397 store.add(createCell(qf1, ts, seqId, value));
1398 store.add(createCell(qf2, ts, seqId, value));
1399 store.add(createCell(qf3, ts, seqId, value));
1400 TreeSet<byte[]> quals = new TreeSet<>(Bytes.BYTES_COMPARATOR);
1401 quals.add(qf1);
1402 quals.add(qf2);
1403 quals.add(qf3);
1404 try (InternalScanner scanner = (InternalScanner) store.getScanner(
1405 new Scan(new Get(row)), quals, seqId)) {
1406 List<Cell> results = new MyList<>(new MyListHook() {
1407 @Override
1408 public void hook(int size) {
1409 switch (size) {
1410
1411
1412
1413 case 1:
1414 try {
1415 flushStore(store, id++);
1416 } catch (IOException e) {
1417 throw new RuntimeException(e);
1418 }
1419 break;
1420
1421
1422 case 2:
1423 try {
1424 byte[] newValue = Bytes.toBytes("newValue");
1425
1426 store.add(createCell(qf1, ts + 1, seqId + 1, newValue));
1427 store.add(createCell(qf2, ts + 1, seqId + 1, newValue));
1428 store.add(createCell(qf3, ts + 1, seqId + 1, newValue));
1429 } catch (IOException e) {
1430 throw new RuntimeException(e);
1431 }
1432 break;
1433 default:
1434 break;
1435 }
1436 }
1437 });
1438 scanner.next(results);
1439 assertEquals(3, results.size());
1440 for (Cell c : results) {
1441 byte[] actualValue = CellUtil.cloneValue(c);
1442 assertTrue("expected:" + Bytes.toStringBinary(value)
1443 + ", actual:" + Bytes.toStringBinary(actualValue)
1444 , Bytes.equals(actualValue, value));
1445 }
1446 }
1447 }
1448
1449 private MyStore initMyStore(String methodName, Configuration conf, MyStoreHook hook) throws IOException {
1450 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(table));
1451 HColumnDescriptor hcd = new HColumnDescriptor(family);
1452 hcd.setMaxVersions(5);
1453 return (MyStore) init(methodName, conf, htd, hcd, hook);
1454 }
1455
1456 private static class MyStore extends HStore {
1457
1458 private final MyStoreHook hook;
1459
1460 MyStore(final HRegion region, final HColumnDescriptor family,
1461 final Configuration confParam, MyStoreHook hook) throws IOException {
1462 super(region, family, confParam);
1463 this.hook = hook;
1464 }
1465
1466 @Override
1467 public List<KeyValueScanner> getScanners(List<StoreFile> files, boolean cacheBlocks,
1468 boolean isGet, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher,
1469 byte[] startRow, byte[] stopRow, long readPt, boolean includeMemstoreScanner) throws IOException {
1470 hook.getScanners(this);
1471 return super.getScanners(files, cacheBlocks, isGet, usePread,
1472 isCompaction, matcher, startRow, stopRow, readPt, includeMemstoreScanner);
1473 }
1474
1475 @Override
1476 public long getSmallestReadPoint() {
1477 return hook.getSmallestReadPoint(this);
1478 }
1479 }
1480
1481 private abstract class MyStoreHook {
1482 void getScanners(MyStore store) throws IOException {
1483 }
1484 long getSmallestReadPoint(HStore store) {
1485 return store.getHRegion().getSmallestReadPoint();
1486 }
1487 }
1488
1489 interface MyListHook {
1490 void hook(int currentSize);
1491 }
1492
1493 private static class MyList<T> implements List<T> {
1494 private final List<T> delegatee = new ArrayList<>();
1495 private final MyListHook hookAtAdd;
1496 MyList(final MyListHook hookAtAdd) {
1497 this.hookAtAdd = hookAtAdd;
1498 }
1499 @Override
1500 public int size() {return delegatee.size();}
1501
1502 @Override
1503 public boolean isEmpty() {return delegatee.isEmpty();}
1504
1505 @Override
1506 public boolean contains(Object o) {return delegatee.contains(o);}
1507
1508 @Override
1509 public Iterator<T> iterator() {return delegatee.iterator();}
1510
1511 @Override
1512 public Object[] toArray() {return delegatee.toArray();}
1513
1514 @Override
1515 public <T> T[] toArray(T[] a) {return delegatee.toArray(a);}
1516
1517 @Override
1518 public boolean add(T e) {
1519 hookAtAdd.hook(size());
1520 return delegatee.add(e);
1521 }
1522
1523 @Override
1524 public boolean remove(Object o) {return delegatee.remove(o);}
1525
1526 @Override
1527 public boolean containsAll(Collection<?> c) {return delegatee.containsAll(c);}
1528
1529 @Override
1530 public boolean addAll(Collection<? extends T> c) {return delegatee.addAll(c);}
1531
1532 @Override
1533 public boolean addAll(int index, Collection<? extends T> c) {return delegatee.addAll(index, c);}
1534
1535 @Override
1536 public boolean removeAll(Collection<?> c) {return delegatee.removeAll(c);}
1537
1538 @Override
1539 public boolean retainAll(Collection<?> c) {return delegatee.retainAll(c);}
1540
1541 @Override
1542 public void clear() {delegatee.clear();}
1543
1544 @Override
1545 public T get(int index) {return delegatee.get(index);}
1546
1547 @Override
1548 public T set(int index, T element) {return delegatee.set(index, element);}
1549
1550 @Override
1551 public void add(int index, T element) {delegatee.add(index, element);}
1552
1553 @Override
1554 public T remove(int index) {return delegatee.remove(index);}
1555
1556 @Override
1557 public int indexOf(Object o) {return delegatee.indexOf(o);}
1558
1559 @Override
1560 public int lastIndexOf(Object o) {return delegatee.lastIndexOf(o);}
1561
1562 @Override
1563 public ListIterator<T> listIterator() {return delegatee.listIterator();}
1564
1565 @Override
1566 public ListIterator<T> listIterator(int index) {return delegatee.listIterator(index);}
1567
1568 @Override
1569 public List<T> subList(int fromIndex, int toIndex) {return delegatee.subList(fromIndex, toIndex);}
1570 }
1571 }