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