View Javadoc

1   /*
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import java.io.IOException;
22  import java.lang.management.ManagementFactory;
23  import java.lang.management.MemoryMXBean;
24  import java.util.ArrayList;
25  import java.util.Arrays;
26  import java.util.List;
27  import java.util.concurrent.atomic.AtomicLong;
28  import java.util.concurrent.atomic.AtomicReference;
29  
30  import junit.framework.TestCase;
31  
32  import org.apache.commons.logging.Log;
33  import org.apache.commons.logging.LogFactory;
34  import org.apache.hadoop.conf.Configuration;
35  import org.apache.hadoop.fs.Path;
36  import org.apache.hadoop.hbase.Cell;
37  import org.apache.hadoop.hbase.CellUtil;
38  import org.apache.hadoop.hbase.HBaseConfiguration;
39  import org.apache.hadoop.hbase.HBaseTestingUtility;
40  import org.apache.hadoop.hbase.HColumnDescriptor;
41  import org.apache.hadoop.hbase.HConstants;
42  import org.apache.hadoop.hbase.HRegionInfo;
43  import org.apache.hadoop.hbase.HTableDescriptor;
44  import org.apache.hadoop.hbase.KeepDeletedCells;
45  import org.apache.hadoop.hbase.KeyValue;
46  import org.apache.hadoop.hbase.KeyValueTestUtil;
47  import org.apache.hadoop.hbase.KeyValueUtil;
48  import org.apache.hadoop.hbase.TableName;
49  import org.apache.hadoop.hbase.client.Scan;
50  import org.apache.hadoop.hbase.testclassification.MediumTests;
51  import org.apache.hadoop.hbase.util.Bytes;
52  import org.apache.hadoop.hbase.util.EnvironmentEdge;
53  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
54  import org.apache.hadoop.hbase.wal.WALFactory;
55  import org.junit.experimental.categories.Category;
56  
57  import com.google.common.base.Joiner;
58  import com.google.common.collect.Iterables;
59  import com.google.common.collect.Lists;
60  
61  import static org.mockito.Mockito.mock;
62  import static org.mockito.Mockito.when;
63  
64  /** memstore test case */
65  @Category(MediumTests.class)
66  public class TestDefaultMemStore extends TestCase {
67    private static final Log LOG = LogFactory.getLog(TestDefaultMemStore.class);
68    private DefaultMemStore memstore;
69    private static final int ROW_COUNT = 10;
70    private static final int QUALIFIER_COUNT = ROW_COUNT;
71    private static final byte [] FAMILY = Bytes.toBytes("column");
72    private MultiVersionConcurrencyControl mvcc;
73    private AtomicLong startSeqNum = new AtomicLong(0); 
74  
75    @Override
76    public void setUp() throws Exception {
77      super.setUp();
78      this.mvcc = new MultiVersionConcurrencyControl();
79      this.memstore = new DefaultMemStore();
80    }
81  
82    public void testPutSameKey() {
83      byte [] bytes = Bytes.toBytes(getName());
84      KeyValue kv = new KeyValue(bytes, bytes, bytes, bytes);
85      this.memstore.add(kv);
86      byte [] other = Bytes.toBytes("somethingelse");
87      KeyValue samekey = new KeyValue(bytes, bytes, bytes, other);
88      this.memstore.add(samekey);
89      Cell found = this.memstore.activeSection.getCellSkipListSet().first();
90      assertEquals(1, this.memstore.activeSection.getCellSkipListSet().sizeForTests());
91      assertTrue(Bytes.toString(found.getValue()), CellUtil.matchingValue(samekey, found));
92    }
93  
94    public void testPutSameCell() {
95      byte[] bytes = Bytes.toBytes(getName());
96      KeyValue kv = new KeyValue(bytes, bytes, bytes, bytes);
97      long sizeChangeForFirstCell = this.memstore.add(kv);
98      long sizeChangeForSecondCell = this.memstore.add(kv);
99      // make sure memstore size increase won't double-count MSLAB chunk size
100     assertEquals(DefaultMemStore.heapSizeChange(kv, true), sizeChangeForFirstCell);
101     if (this.memstore.activeSection.getMemStoreLAB() != null) {
102       // make sure memstore size increased when using MSLAB
103       assertEquals(memstore.getCellLength(kv), sizeChangeForSecondCell);
104       // make sure chunk size increased even when writing the same cell, if using MSLAB
105       if (this.memstore.activeSection.getMemStoreLAB() instanceof HeapMemStoreLAB) {
106         assertEquals(2 * memstore.getCellLength(kv),
107           ((HeapMemStoreLAB) this.memstore.activeSection.getMemStoreLAB())
108             .getCurrentChunk().getNextFreeOffset());
109       }
110     } else {
111       // make sure no memstore size change w/o MSLAB
112       assertEquals(0, sizeChangeForSecondCell);
113     }
114   }
115 
116   /**
117    * Test memstore snapshot happening while scanning.
118    * @throws IOException
119    */
120   public void testScanAcrossSnapshot() throws IOException {
121     int rowCount = addRows(this.memstore);
122     List<KeyValueScanner> memstorescanners = this.memstore.getScanners(0);
123     Scan scan = new Scan();
124     List<Cell> result = new ArrayList<Cell>();
125     Configuration conf = HBaseConfiguration.create();
126     ScanInfo scanInfo =
127         new ScanInfo(conf, null, 0, 1, HConstants.LATEST_TIMESTAMP, KeepDeletedCells.FALSE, 0,
128             this.memstore.comparator);
129     ScanType scanType = ScanType.USER_SCAN;
130     StoreScanner s = new StoreScanner(scan, scanInfo, scanType, null, memstorescanners);
131     int count = 0;
132     try {
133       while (s.next(result)) {
134         LOG.info(result);
135         count++;
136         // Row count is same as column count.
137         assertEquals(rowCount, result.size());
138         result.clear();
139       }
140     } finally {
141       s.close();
142     }
143     assertEquals(rowCount, count);
144     for (KeyValueScanner scanner : memstorescanners) {
145       scanner.close();
146     }
147 
148     memstorescanners = this.memstore.getScanners(mvcc.getReadPoint());
149     // Now assert can count same number even if a snapshot mid-scan.
150     s = new StoreScanner(scan, scanInfo, scanType, null, memstorescanners);
151     count = 0;
152     try {
153       while (s.next(result)) {
154         LOG.info(result);
155         // Assert the stuff is coming out in right order.
156         assertTrue(CellUtil.matchingRow(result.get(0), Bytes.toBytes(count)));
157         count++;
158         // Row count is same as column count.
159         assertEquals(rowCount, result.size());
160         if (count == 2) {
161           this.memstore.snapshot();
162           LOG.info("Snapshotted");
163         }
164         result.clear();
165       }
166     } finally {
167       s.close();
168     }
169     assertEquals(rowCount, count);
170     for (KeyValueScanner scanner : memstorescanners) {
171       scanner.close();
172     }
173     memstorescanners = this.memstore.getScanners(mvcc.getReadPoint());
174     // Assert that new values are seen in kvset as we scan.
175     long ts = System.currentTimeMillis();
176     s = new StoreScanner(scan, scanInfo, scanType, null, memstorescanners);
177     count = 0;
178     int snapshotIndex = 5;
179     try {
180       while (s.next(result)) {
181         LOG.info(result);
182         // Assert the stuff is coming out in right order.
183         assertTrue(CellUtil.matchingRow(result.get(0), Bytes.toBytes(count)));
184         // Row count is same as column count.
185         assertEquals("count=" + count + ", result=" + result, rowCount, result.size());
186         count++;
187         if (count == snapshotIndex) {
188           MemStoreSnapshot snapshot = this.memstore.snapshot();
189           this.memstore.clearSnapshot(snapshot.getId());
190           // Added more rows into kvset.  But the scanner wont see these rows.
191           addRows(this.memstore, ts);
192           LOG.info("Snapshotted, cleared it and then added values (which wont be seen)");
193         }
194         result.clear();
195       }
196     } finally {
197       s.close();
198     }
199     assertEquals(rowCount, count);
200   }
201 
202   /**
203    * A simple test which verifies the 3 possible states when scanning across snapshot.
204    * @throws IOException
205    * @throws CloneNotSupportedException 
206    */
207   public void testScanAcrossSnapshot2() throws IOException, CloneNotSupportedException {
208     // we are going to the scanning across snapshot with two kvs
209     // kv1 should always be returned before kv2
210     final byte[] one = Bytes.toBytes(1);
211     final byte[] two = Bytes.toBytes(2);
212     final byte[] f = Bytes.toBytes("f");
213     final byte[] q = Bytes.toBytes("q");
214     final byte[] v = Bytes.toBytes(3);
215 
216     final KeyValue kv1 = new KeyValue(one, f, q, v);
217     final KeyValue kv2 = new KeyValue(two, f, q, v);
218 
219     // use case 1: both kvs in kvset
220     this.memstore.add(kv1.clone());
221     this.memstore.add(kv2.clone());
222     verifyScanAcrossSnapshot2(kv1, kv2);
223 
224     // use case 2: both kvs in snapshot
225     this.memstore.snapshot();
226     verifyScanAcrossSnapshot2(kv1, kv2);
227 
228     // use case 3: first in snapshot second in kvset
229     this.memstore = new DefaultMemStore();
230     this.memstore.add(kv1.clone());
231     this.memstore.snapshot();
232     this.memstore.add(kv2.clone());
233     verifyScanAcrossSnapshot2(kv1, kv2);
234   }
235 
236   private void verifyScanAcrossSnapshot2(KeyValue kv1, KeyValue kv2)
237       throws IOException {
238     List<KeyValueScanner> memstorescanners = this.memstore.getScanners(mvcc.getReadPoint());
239     assertEquals(1, memstorescanners.size());
240     final KeyValueScanner scanner = memstorescanners.get(0);
241     scanner.seek(KeyValueUtil.createFirstOnRow(HConstants.EMPTY_START_ROW));
242     assertEquals(kv1, scanner.next());
243     assertEquals(kv2, scanner.next());
244     assertNull(scanner.next());
245   }
246 
247   private void assertScannerResults(KeyValueScanner scanner, KeyValue[] expected)
248       throws IOException {
249     scanner.seek(KeyValueUtil.createFirstOnRow(new byte[]{}));
250     List<Cell> returned = Lists.newArrayList();
251 
252     while (true) {
253       Cell next = scanner.next();
254       if (next == null) break;
255       returned.add(next);
256     }
257 
258     assertTrue(
259         "Got:\n" + Joiner.on("\n").join(returned) +
260         "\nExpected:\n" + Joiner.on("\n").join(expected),
261         Iterables.elementsEqual(Arrays.asList(expected), returned));
262     assertNull(scanner.peek());
263   }
264 
265   public void testMemstoreConcurrentControl() throws IOException {
266     final byte[] row = Bytes.toBytes(1);
267     final byte[] f = Bytes.toBytes("family");
268     final byte[] q1 = Bytes.toBytes("q1");
269     final byte[] q2 = Bytes.toBytes("q2");
270     final byte[] v = Bytes.toBytes("value");
271 
272     MultiVersionConcurrencyControl.WriteEntry w =
273         mvcc.begin();
274 
275     KeyValue kv1 = new KeyValue(row, f, q1, v);
276     kv1.setSequenceId(w.getWriteNumber());
277     memstore.add(kv1);
278 
279     assertTrue(this.memstore.getScanners(mvcc.getReadPoint()) == null);
280 
281     mvcc.completeAndWait(w);
282 
283     KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
284     assertScannerResults(s, new KeyValue[]{kv1});
285 
286     w = mvcc.begin();
287     KeyValue kv2 = new KeyValue(row, f, q2, v);
288     kv2.setSequenceId(w.getWriteNumber());
289     memstore.add(kv2);
290 
291     s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
292     assertScannerResults(s, new KeyValue[]{kv1});
293 
294     mvcc.completeAndWait(w);
295 
296     s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
297     assertScannerResults(s, new KeyValue[]{kv1, kv2});
298   }
299 
300   /**
301    * Regression test for HBASE-2616, HBASE-2670.
302    * When we insert a higher-memstoreTS version of a cell but with
303    * the same timestamp, we still need to provide consistent reads
304    * for the same scanner.
305    */
306   public void testMemstoreEditsVisibilityWithSameKey() throws IOException {
307     final byte[] row = Bytes.toBytes(1);
308     final byte[] f = Bytes.toBytes("family");
309     final byte[] q1 = Bytes.toBytes("q1");
310     final byte[] q2 = Bytes.toBytes("q2");
311     final byte[] v1 = Bytes.toBytes("value1");
312     final byte[] v2 = Bytes.toBytes("value2");
313 
314     // INSERT 1: Write both columns val1
315     MultiVersionConcurrencyControl.WriteEntry w =
316         mvcc.begin();
317 
318     KeyValue kv11 = new KeyValue(row, f, q1, v1);
319     kv11.setSequenceId(w.getWriteNumber());
320     memstore.add(kv11);
321 
322     KeyValue kv12 = new KeyValue(row, f, q2, v1);
323     kv12.setSequenceId(w.getWriteNumber());
324     memstore.add(kv12);
325     mvcc.completeAndWait(w);
326 
327     // BEFORE STARTING INSERT 2, SEE FIRST KVS
328     KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
329     assertScannerResults(s, new KeyValue[]{kv11, kv12});
330 
331     // START INSERT 2: Write both columns val2
332     w = mvcc.begin();
333     KeyValue kv21 = new KeyValue(row, f, q1, v2);
334     kv21.setSequenceId(w.getWriteNumber());
335     memstore.add(kv21);
336 
337     KeyValue kv22 = new KeyValue(row, f, q2, v2);
338     kv22.setSequenceId(w.getWriteNumber());
339     memstore.add(kv22);
340 
341     // BEFORE COMPLETING INSERT 2, SEE FIRST KVS
342     s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
343     assertScannerResults(s, new KeyValue[]{kv11, kv12});
344 
345     // COMPLETE INSERT 2
346     mvcc.completeAndWait(w);
347 
348     // NOW SHOULD SEE NEW KVS IN ADDITION TO OLD KVS.
349     // See HBASE-1485 for discussion about what we should do with
350     // the duplicate-TS inserts
351     s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
352     assertScannerResults(s, new KeyValue[]{kv21, kv11, kv22, kv12});
353   }
354 
355   /**
356    * When we insert a higher-memstoreTS deletion of a cell but with
357    * the same timestamp, we still need to provide consistent reads
358    * for the same scanner.
359    */
360   public void testMemstoreDeletesVisibilityWithSameKey() throws IOException {
361     final byte[] row = Bytes.toBytes(1);
362     final byte[] f = Bytes.toBytes("family");
363     final byte[] q1 = Bytes.toBytes("q1");
364     final byte[] q2 = Bytes.toBytes("q2");
365     final byte[] v1 = Bytes.toBytes("value1");
366     // INSERT 1: Write both columns val1
367     MultiVersionConcurrencyControl.WriteEntry w =
368         mvcc.begin();
369 
370     KeyValue kv11 = new KeyValue(row, f, q1, v1);
371     kv11.setSequenceId(w.getWriteNumber());
372     memstore.add(kv11);
373 
374     KeyValue kv12 = new KeyValue(row, f, q2, v1);
375     kv12.setSequenceId(w.getWriteNumber());
376     memstore.add(kv12);
377     mvcc.completeAndWait(w);
378 
379     // BEFORE STARTING INSERT 2, SEE FIRST KVS
380     KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
381     assertScannerResults(s, new KeyValue[]{kv11, kv12});
382 
383     // START DELETE: Insert delete for one of the columns
384     w = mvcc.begin();
385     KeyValue kvDel = new KeyValue(row, f, q2, kv11.getTimestamp(),
386         KeyValue.Type.DeleteColumn);
387     kvDel.setSequenceId(w.getWriteNumber());
388     memstore.add(kvDel);
389 
390     // BEFORE COMPLETING DELETE, SEE FIRST KVS
391     s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
392     assertScannerResults(s, new KeyValue[]{kv11, kv12});
393 
394     // COMPLETE DELETE
395     mvcc.completeAndWait(w);
396 
397     // NOW WE SHOULD SEE DELETE
398     s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
399     assertScannerResults(s, new KeyValue[]{kv11, kvDel, kv12});
400   }
401 
402 
403   private static class ReadOwnWritesTester extends Thread {
404     static final int NUM_TRIES = 1000;
405 
406     final byte[] row;
407 
408     final byte[] f = Bytes.toBytes("family");
409     final byte[] q1 = Bytes.toBytes("q1");
410 
411     final MultiVersionConcurrencyControl mvcc;
412     final MemStore memstore;
413     final AtomicLong startSeqNum;
414 
415     AtomicReference<Throwable> caughtException;
416 
417 
418     public ReadOwnWritesTester(int id,
419                                MemStore memstore,
420                                MultiVersionConcurrencyControl mvcc,
421                                AtomicReference<Throwable> caughtException,
422                                AtomicLong startSeqNum)
423     {
424       this.mvcc = mvcc;
425       this.memstore = memstore;
426       this.caughtException = caughtException;
427       row = Bytes.toBytes(id);
428       this.startSeqNum = startSeqNum;
429     }
430 
431     public void run() {
432       try {
433         internalRun();
434       } catch (Throwable t) {
435         caughtException.compareAndSet(null, t);
436       }
437     }
438 
439     private void internalRun() throws IOException {
440       for (long i = 0; i < NUM_TRIES && caughtException.get() == null; i++) {
441         MultiVersionConcurrencyControl.WriteEntry w =
442             mvcc.begin();
443 
444         // Insert the sequence value (i)
445         byte[] v = Bytes.toBytes(i);
446 
447         KeyValue kv = new KeyValue(row, f, q1, i, v);
448         kv.setSequenceId(w.getWriteNumber());
449         memstore.add(kv);
450         mvcc.completeAndWait(w);
451 
452         // Assert that we can read back
453         KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
454         s.seek(kv);
455 
456         Cell ret = s.next();
457         assertNotNull("Didnt find own write at all", ret);
458         assertEquals("Didnt read own writes",
459                      kv.getTimestamp(), ret.getTimestamp());
460       }
461     }
462   }
463 
464   public void testReadOwnWritesUnderConcurrency() throws Throwable {
465 
466     int NUM_THREADS = 8;
467 
468     ReadOwnWritesTester threads[] = new ReadOwnWritesTester[NUM_THREADS];
469     AtomicReference<Throwable> caught = new AtomicReference<Throwable>();
470 
471     for (int i = 0; i < NUM_THREADS; i++) {
472       threads[i] = new ReadOwnWritesTester(i, memstore, mvcc, caught, this.startSeqNum);
473       threads[i].start();
474     }
475 
476     for (int i = 0; i < NUM_THREADS; i++) {
477       threads[i].join();
478     }
479 
480     if (caught.get() != null) {
481       throw caught.get();
482     }
483   }
484 
485   /**
486    * Test memstore snapshots
487    * @throws IOException
488    */
489   public void testSnapshotting() throws IOException {
490     final int snapshotCount = 5;
491     // Add some rows, run a snapshot. Do it a few times.
492     for (int i = 0; i < snapshotCount; i++) {
493       addRows(this.memstore);
494       runSnapshot(this.memstore);
495       assertEquals("History not being cleared", 0,
496         this.memstore.snapshotSection.getCellSkipListSet().sizeForTests());
497     }
498   }
499 
500   public void testMultipleVersionsSimple() throws Exception {
501     DefaultMemStore m = new DefaultMemStore(new Configuration(), KeyValue.COMPARATOR);
502     byte [] row = Bytes.toBytes("testRow");
503     byte [] family = Bytes.toBytes("testFamily");
504     byte [] qf = Bytes.toBytes("testQualifier");
505     long [] stamps = {1,2,3};
506     byte [][] values = {Bytes.toBytes("value0"), Bytes.toBytes("value1"),
507         Bytes.toBytes("value2")};
508     KeyValue key0 = new KeyValue(row, family, qf, stamps[0], values[0]);
509     KeyValue key1 = new KeyValue(row, family, qf, stamps[1], values[1]);
510     KeyValue key2 = new KeyValue(row, family, qf, stamps[2], values[2]);
511 
512     m.add(key0);
513     m.add(key1);
514     m.add(key2);
515 
516     assertTrue("Expected memstore to hold 3 values, actually has " +
517         m.activeSection.getCellSkipListSet().sizeForTests(),
518         m.activeSection.getCellSkipListSet().sizeForTests() == 3);
519   }
520 
521   //////////////////////////////////////////////////////////////////////////////
522   // Get tests
523   //////////////////////////////////////////////////////////////////////////////
524 
525   /** Test getNextRow from memstore
526    * @throws InterruptedException
527    */
528   public void testGetNextRow() throws Exception {
529     addRows(this.memstore);
530     // Add more versions to make it a little more interesting.
531     Thread.sleep(1);
532     addRows(this.memstore);
533     Cell closestToEmpty = this.memstore.getNextRow(KeyValue.LOWESTKEY);
534     assertTrue(KeyValue.COMPARATOR.compareRows(closestToEmpty,
535       new KeyValue(Bytes.toBytes(0), System.currentTimeMillis())) == 0);
536     for (int i = 0; i < ROW_COUNT; i++) {
537       Cell nr = this.memstore.getNextRow(new KeyValue(Bytes.toBytes(i),
538         System.currentTimeMillis()));
539       if (i + 1 == ROW_COUNT) {
540         assertEquals(nr, null);
541       } else {
542         assertTrue(KeyValue.COMPARATOR.compareRows(nr,
543           new KeyValue(Bytes.toBytes(i + 1), System.currentTimeMillis())) == 0);
544       }
545     }
546     //starting from each row, validate results should contain the starting row
547     Configuration conf = HBaseConfiguration.create();
548     for (int startRowId = 0; startRowId < ROW_COUNT; startRowId++) {
549       ScanInfo scanInfo = new ScanInfo(conf, FAMILY, 0, 1, Integer.MAX_VALUE,
550         KeepDeletedCells.FALSE, 0, this.memstore.comparator);
551       ScanType scanType = ScanType.USER_SCAN;
552       InternalScanner scanner = new StoreScanner(new Scan(
553           Bytes.toBytes(startRowId)), scanInfo, scanType, null,
554           memstore.getScanners(0));
555       List<Cell> results = new ArrayList<Cell>();
556       for (int i = 0; scanner.next(results); i++) {
557         int rowId = startRowId + i;
558         Cell left = results.get(0);
559         byte[] row1 = Bytes.toBytes(rowId);
560         assertTrue(
561             "Row name",
562             KeyValue.COMPARATOR.compareRows(left.getRowArray(), left.getRowOffset(),
563                 (int) left.getRowLength(), row1, 0, row1.length) == 0);
564         assertEquals("Count of columns", QUALIFIER_COUNT, results.size());
565         List<Cell> row = new ArrayList<Cell>();
566         for (Cell kv : results) {
567           row.add(kv);
568         }
569         isExpectedRowWithoutTimestamps(rowId, row);
570         // Clear out set.  Otherwise row results accumulate.
571         results.clear();
572       }
573     }
574   }
575 
576   public void testGet_memstoreAndSnapShot() throws IOException {
577     byte [] row = Bytes.toBytes("testrow");
578     byte [] fam = Bytes.toBytes("testfamily");
579     byte [] qf1 = Bytes.toBytes("testqualifier1");
580     byte [] qf2 = Bytes.toBytes("testqualifier2");
581     byte [] qf3 = Bytes.toBytes("testqualifier3");
582     byte [] qf4 = Bytes.toBytes("testqualifier4");
583     byte [] qf5 = Bytes.toBytes("testqualifier5");
584     byte [] val = Bytes.toBytes("testval");
585 
586     //Setting up memstore
587     memstore.add(new KeyValue(row, fam ,qf1, val));
588     memstore.add(new KeyValue(row, fam ,qf2, val));
589     memstore.add(new KeyValue(row, fam ,qf3, val));
590     //Creating a snapshot
591     memstore.snapshot();
592     assertEquals(3, memstore.snapshotSection.getCellSkipListSet().sizeForTests());
593     //Adding value to "new" memstore
594     assertEquals(0, memstore.activeSection.getCellSkipListSet().sizeForTests());
595     memstore.add(new KeyValue(row, fam ,qf4, val));
596     memstore.add(new KeyValue(row, fam ,qf5, val));
597     assertEquals(2, memstore.activeSection.getCellSkipListSet().sizeForTests());
598   }
599 
600   //////////////////////////////////////////////////////////////////////////////
601   // Delete tests
602   //////////////////////////////////////////////////////////////////////////////
603   public void testGetWithDelete() throws IOException {
604     byte [] row = Bytes.toBytes("testrow");
605     byte [] fam = Bytes.toBytes("testfamily");
606     byte [] qf1 = Bytes.toBytes("testqualifier");
607     byte [] val = Bytes.toBytes("testval");
608 
609     long ts1 = System.nanoTime();
610     KeyValue put1 = new KeyValue(row, fam, qf1, ts1, val);
611     long ts2 = ts1 + 1;
612     KeyValue put2 = new KeyValue(row, fam, qf1, ts2, val);
613     long ts3 = ts2 +1;
614     KeyValue put3 = new KeyValue(row, fam, qf1, ts3, val);
615     memstore.add(put1);
616     memstore.add(put2);
617     memstore.add(put3);
618 
619     assertEquals(3, memstore.activeSection.getCellSkipListSet().sizeForTests());
620 
621     KeyValue del2 = new KeyValue(row, fam, qf1, ts2, KeyValue.Type.Delete, val);
622     memstore.delete(del2);
623 
624     List<Cell> expected = new ArrayList<Cell>();
625     expected.add(put3);
626     expected.add(del2);
627     expected.add(put2);
628     expected.add(put1);
629 
630     assertEquals(4, memstore.activeSection.getCellSkipListSet().sizeForTests());
631     int i = 0;
632     for(Cell cell : memstore.activeSection.getCellSkipListSet()) {
633       assertEquals(expected.get(i++), cell);
634     }
635   }
636 
637   public void testGetWithDeleteColumn() throws IOException {
638     byte [] row = Bytes.toBytes("testrow");
639     byte [] fam = Bytes.toBytes("testfamily");
640     byte [] qf1 = Bytes.toBytes("testqualifier");
641     byte [] val = Bytes.toBytes("testval");
642 
643     long ts1 = System.nanoTime();
644     KeyValue put1 = new KeyValue(row, fam, qf1, ts1, val);
645     long ts2 = ts1 + 1;
646     KeyValue put2 = new KeyValue(row, fam, qf1, ts2, val);
647     long ts3 = ts2 +1;
648     KeyValue put3 = new KeyValue(row, fam, qf1, ts3, val);
649     memstore.add(put1);
650     memstore.add(put2);
651     memstore.add(put3);
652 
653     assertEquals(3, memstore.activeSection.getCellSkipListSet().sizeForTests());
654 
655     KeyValue del2 =
656       new KeyValue(row, fam, qf1, ts2, KeyValue.Type.DeleteColumn, val);
657     memstore.delete(del2);
658 
659     List<Cell> expected = new ArrayList<Cell>();
660     expected.add(put3);
661     expected.add(del2);
662     expected.add(put2);
663     expected.add(put1);
664 
665     assertEquals(4, memstore.activeSection.getCellSkipListSet().sizeForTests());
666     int i = 0;
667     for (Cell cell: memstore.activeSection.getCellSkipListSet()) {
668       assertEquals(expected.get(i++), cell);
669     }
670   }
671 
672   public void testGetWithDeleteFamily() throws IOException {
673     byte [] row = Bytes.toBytes("testrow");
674     byte [] fam = Bytes.toBytes("testfamily");
675     byte [] qf1 = Bytes.toBytes("testqualifier1");
676     byte [] qf2 = Bytes.toBytes("testqualifier2");
677     byte [] qf3 = Bytes.toBytes("testqualifier3");
678     byte [] val = Bytes.toBytes("testval");
679     long ts = System.nanoTime();
680 
681     KeyValue put1 = new KeyValue(row, fam, qf1, ts, val);
682     KeyValue put2 = new KeyValue(row, fam, qf2, ts, val);
683     KeyValue put3 = new KeyValue(row, fam, qf3, ts, val);
684     KeyValue put4 = new KeyValue(row, fam, qf3, ts+1, val);
685 
686     memstore.add(put1);
687     memstore.add(put2);
688     memstore.add(put3);
689     memstore.add(put4);
690 
691     KeyValue del =
692       new KeyValue(row, fam, null, ts, KeyValue.Type.DeleteFamily, val);
693     memstore.delete(del);
694 
695     List<Cell> expected = new ArrayList<Cell>();
696     expected.add(del);
697     expected.add(put1);
698     expected.add(put2);
699     expected.add(put4);
700     expected.add(put3);
701 
702     assertEquals(5, memstore.activeSection.getCellSkipListSet().sizeForTests());
703     int i = 0;
704     for (Cell cell: memstore.activeSection.getCellSkipListSet()) {
705       assertEquals(expected.get(i++), cell);
706     }
707   }
708 
709   public void testKeepDeleteInmemstore() {
710     byte [] row = Bytes.toBytes("testrow");
711     byte [] fam = Bytes.toBytes("testfamily");
712     byte [] qf = Bytes.toBytes("testqualifier");
713     byte [] val = Bytes.toBytes("testval");
714     long ts = System.nanoTime();
715     memstore.add(new KeyValue(row, fam, qf, ts, val));
716     KeyValue delete = new KeyValue(row, fam, qf, ts, KeyValue.Type.Delete, val);
717     memstore.delete(delete);
718     assertEquals(2, memstore.activeSection.getCellSkipListSet().sizeForTests());
719     assertEquals(delete, memstore.activeSection.getCellSkipListSet().first());
720   }
721 
722   public void testRetainsDeleteVersion() throws IOException {
723     // add a put to memstore
724     memstore.add(KeyValueTestUtil.create("row1", "fam", "a", 100, "dont-care"));
725 
726     // now process a specific delete:
727     KeyValue delete = KeyValueTestUtil.create(
728         "row1", "fam", "a", 100, KeyValue.Type.Delete, "dont-care");
729     memstore.delete(delete);
730 
731     assertEquals(2, memstore.activeSection.getCellSkipListSet().sizeForTests());
732     assertEquals(delete, memstore.activeSection.getCellSkipListSet().first());
733   }
734   public void testRetainsDeleteColumn() throws IOException {
735     // add a put to memstore
736     memstore.add(KeyValueTestUtil.create("row1", "fam", "a", 100, "dont-care"));
737 
738     // now process a specific delete:
739     KeyValue delete = KeyValueTestUtil.create("row1", "fam", "a", 100,
740         KeyValue.Type.DeleteColumn, "dont-care");
741     memstore.delete(delete);
742 
743     assertEquals(2, memstore.activeSection.getCellSkipListSet().sizeForTests());
744     assertEquals(delete, memstore.activeSection.getCellSkipListSet().first());
745   }
746 
747   public void testRetainsDeleteFamily() throws IOException {
748     // add a put to memstore
749     memstore.add(KeyValueTestUtil.create("row1", "fam", "a", 100, "dont-care"));
750 
751     // now process a specific delete:
752     KeyValue delete = KeyValueTestUtil.create("row1", "fam", "a", 100,
753         KeyValue.Type.DeleteFamily, "dont-care");
754     memstore.delete(delete);
755 
756     assertEquals(2, memstore.activeSection.getCellSkipListSet().sizeForTests());
757     assertEquals(delete, memstore.activeSection.getCellSkipListSet().first());
758   }
759 
760   ////////////////////////////////////
761   //Test for timestamps
762   ////////////////////////////////////
763 
764   /**
765    * Test to ensure correctness when using Memstore with multiple timestamps
766    */
767   public void testMultipleTimestamps() throws Exception {
768     long[] timestamps = new long[] {20,10,5,1};
769     Scan scan = new Scan();
770 
771     for (long timestamp: timestamps)
772       addRows(memstore,timestamp);
773 
774     byte[] fam = Bytes.toBytes("fam");
775     HColumnDescriptor hcd = mock(HColumnDescriptor.class);
776     when(hcd.getName()).thenReturn(fam);
777     Store store = mock(Store.class);
778     when(store.getFamily()).thenReturn(hcd);
779     scan.setColumnFamilyTimeRange(fam, 0, 2);
780     assertTrue(memstore.shouldSeek(scan, store, Long.MIN_VALUE));
781 
782     scan.setColumnFamilyTimeRange(fam, 20, 82);
783     assertTrue(memstore.shouldSeek(scan, store, Long.MIN_VALUE));
784 
785     scan.setColumnFamilyTimeRange(fam, 10, 20);
786     assertTrue(memstore.shouldSeek(scan, store, Long.MIN_VALUE));
787 
788     scan.setColumnFamilyTimeRange(fam, 8, 12);
789     assertTrue(memstore.shouldSeek(scan, store, Long.MIN_VALUE));
790 
791     scan.setColumnFamilyTimeRange(fam, 28, 42);
792     assertTrue(!memstore.shouldSeek(scan, store, Long.MIN_VALUE));
793   }
794 
795   ////////////////////////////////////
796   //Test for upsert with MSLAB
797   ////////////////////////////////////
798 
799   /**
800    * Test a pathological pattern that shows why we can't currently
801    * use the MSLAB for upsert workloads. This test inserts data
802    * in the following pattern:
803    *
804    * - row0001 through row1000 (fills up one 2M Chunk)
805    * - row0002 through row1001 (fills up another 2M chunk, leaves one reference
806    *   to the first chunk
807    * - row0003 through row1002 (another chunk, another dangling reference)
808    *
809    * This causes OOME pretty quickly if we use MSLAB for upsert
810    * since each 2M chunk is held onto by a single reference.
811    */
812   public void testUpsertMSLAB() throws Exception {
813     Configuration conf = HBaseConfiguration.create();
814     conf.setBoolean(DefaultMemStore.USEMSLAB_KEY, true);
815     memstore = new DefaultMemStore(conf, KeyValue.COMPARATOR);
816 
817     int ROW_SIZE = 2048;
818     byte[] qualifier = new byte[ROW_SIZE - 4];
819 
820     MemoryMXBean bean = ManagementFactory.getMemoryMXBean();
821     for (int i = 0; i < 3; i++) { System.gc(); }
822     long usageBefore = bean.getHeapMemoryUsage().getUsed();
823 
824     long size = 0;
825     long ts=0;
826 
827     for (int newValue = 0; newValue < 1000; newValue++) {
828       for (int row = newValue; row < newValue + 1000; row++) {
829         byte[] rowBytes = Bytes.toBytes(row);
830         size += memstore.updateColumnValue(rowBytes, FAMILY, qualifier, newValue, ++ts);
831       }
832     }
833     System.out.println("Wrote " + ts + " vals");
834     for (int i = 0; i < 3; i++) { System.gc(); }
835     long usageAfter = bean.getHeapMemoryUsage().getUsed();
836     System.out.println("Memory used: " + (usageAfter - usageBefore)
837         + " (heapsize: " + memstore.heapSize() +
838         " size: " + size + ")");
839   }
840 
841   //////////////////////////////////////////////////////////////////////////////
842   // Helpers
843   //////////////////////////////////////////////////////////////////////////////
844   private static byte [] makeQualifier(final int i1, final int i2){
845     return Bytes.toBytes(Integer.toString(i1) + ";" +
846         Integer.toString(i2));
847   }
848 
849   /**
850    * Add keyvalues with a fixed memstoreTs, and checks that memstore size is decreased
851    * as older keyvalues are deleted from the memstore.
852    * @throws Exception
853    */
854   public void testUpsertMemstoreSize() throws Exception {
855     Configuration conf = HBaseConfiguration.create();
856     memstore = new DefaultMemStore(conf, KeyValue.COMPARATOR);
857     long oldSize = memstore.activeSection.getHeapSize().get();
858 
859     List<Cell> l = new ArrayList<Cell>();
860     KeyValue kv1 = KeyValueTestUtil.create("r", "f", "q", 100, "v");
861     KeyValue kv2 = KeyValueTestUtil.create("r", "f", "q", 101, "v");
862     KeyValue kv3 = KeyValueTestUtil.create("r", "f", "q", 102, "v");
863 
864     kv1.setSequenceId(1); kv2.setSequenceId(1);kv3.setSequenceId(1);
865     l.add(kv1); l.add(kv2); l.add(kv3);
866 
867     this.memstore.upsert(l, 2, null);// readpoint is 2
868     long newSize = this.memstore.activeSection.getHeapSize().get();
869     assert(newSize > oldSize);
870     //The kv1 should be removed.
871     assert(memstore.activeSection.getCellSkipListSet().sizeForTests() == 2);
872 
873     KeyValue kv4 = KeyValueTestUtil.create("r", "f", "q", 104, "v");
874     kv4.setSequenceId(1);
875     l.clear(); l.add(kv4);
876     this.memstore.upsert(l, 3, null);
877     assertEquals(newSize, this.memstore.activeSection.getHeapSize().get());
878     //The kv2 should be removed.
879     assert(memstore.activeSection.getCellSkipListSet().sizeForTests() == 2);
880     //this.memstore = null;
881   }
882 
883   ////////////////////////////////////
884   // Test for periodic memstore flushes 
885   // based on time of oldest edit
886   ////////////////////////////////////
887 
888   /**
889    * Tests that the timeOfOldestEdit is updated correctly for the 
890    * various edit operations in memstore.
891    * @throws Exception
892    */
893   public void testUpdateToTimeOfOldestEdit() throws Exception {
894     try {
895       EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
896       EnvironmentEdgeManager.injectEdge(edge);
897       DefaultMemStore memstore = new DefaultMemStore();
898       long t = memstore.timeOfOldestEdit();
899       assertEquals(t, Long.MAX_VALUE);
900 
901       // test the case that the timeOfOldestEdit is updated after a KV add
902       memstore.add(KeyValueTestUtil.create("r", "f", "q", 100, "v"));
903       t = memstore.timeOfOldestEdit();
904       assertTrue(t == 1234);
905       // snapshot() will reset timeOfOldestEdit. The method will also assert the 
906       // value is reset to Long.MAX_VALUE
907       t = runSnapshot(memstore);
908 
909       // test the case that the timeOfOldestEdit is updated after a KV delete
910       memstore.delete(KeyValueTestUtil.create("r", "f", "q", 100, "v"));
911       t = memstore.timeOfOldestEdit();
912       assertTrue(t == 1234);
913       t = runSnapshot(memstore);
914 
915       // test the case that the timeOfOldestEdit is updated after a KV upsert
916       List<Cell> l = new ArrayList<Cell>();
917       KeyValue kv1 = KeyValueTestUtil.create("r", "f", "q", 100, "v");
918       kv1.setSequenceId(100);
919       l.add(kv1);
920       memstore.upsert(l, 1000, null);
921       t = memstore.timeOfOldestEdit();
922       assertTrue(t == 1234);
923     } finally {
924       EnvironmentEdgeManager.reset();
925     }
926   }
927 
928   /**
929    * Tests the HRegion.shouldFlush method - adds an edit in the memstore
930    * and checks that shouldFlush returns true, and another where it disables
931    * the periodic flush functionality and tests whether shouldFlush returns
932    * false. 
933    * @throws Exception
934    */
935   public void testShouldFlush() throws Exception {
936     Configuration conf = new Configuration();
937     conf.setInt(HRegion.MEMSTORE_PERIODIC_FLUSH_INTERVAL, 1000);
938     checkShouldFlush(conf, true);
939     // test disable flush
940     conf.setInt(HRegion.MEMSTORE_PERIODIC_FLUSH_INTERVAL, 0);
941     checkShouldFlush(conf, false);
942   }
943 
944   private void checkShouldFlush(Configuration conf, boolean expected) throws Exception {
945     try {
946       EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
947       EnvironmentEdgeManager.injectEdge(edge);
948       HBaseTestingUtility hbaseUtility = HBaseTestingUtility.createLocalHTU(conf);
949       HRegion region = hbaseUtility.createTestRegion("foobar", new HColumnDescriptor("foo"));
950 
951       List<Store> stores = region.getStores();
952       assertTrue(stores.size() == 1);
953 
954       Store s = stores.iterator().next();
955       edge.setCurrentTimeMillis(1234);
956       s.add(KeyValueTestUtil.create("r", "f", "q", 100, "v"));
957       edge.setCurrentTimeMillis(1234 + 100);
958       StringBuffer sb = new StringBuffer();
959       assertTrue(region.shouldFlush(sb) == false);
960       edge.setCurrentTimeMillis(1234 + 10000);
961       assertTrue(region.shouldFlush(sb) == expected);
962     } finally {
963       EnvironmentEdgeManager.reset();
964     }
965   }
966 
967   public void testShouldFlushMeta() throws Exception {
968     // write an edit in the META and ensure the shouldFlush (that the periodic memstore
969     // flusher invokes) returns true after SYSTEM_CACHE_FLUSH_INTERVAL (even though
970     // the MEMSTORE_PERIODIC_FLUSH_INTERVAL is set to a higher value)
971     Configuration conf = new Configuration();
972     conf.setInt(HRegion.MEMSTORE_PERIODIC_FLUSH_INTERVAL, HRegion.SYSTEM_CACHE_FLUSH_INTERVAL * 10);
973     HBaseTestingUtility hbaseUtility = HBaseTestingUtility.createLocalHTU(conf);
974     Path testDir = hbaseUtility.getDataTestDir();
975     EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
976     EnvironmentEdgeManager.injectEdge(edge);
977     edge.setCurrentTimeMillis(1234);
978     WALFactory wFactory = new WALFactory(conf, null, "1234");
979     HRegion meta = HRegion.createHRegion(HRegionInfo.FIRST_META_REGIONINFO, testDir,
980         conf, HTableDescriptor.metaTableDescriptor(conf),
981         wFactory.getMetaWAL(HRegionInfo.FIRST_META_REGIONINFO.
982             getEncodedNameAsBytes()));
983     HRegionInfo hri = new HRegionInfo(TableName.valueOf("testShouldFlushMeta"),
984         Bytes.toBytes("row_0200"), Bytes.toBytes("row_0300"));
985     HTableDescriptor desc = new HTableDescriptor(TableName.valueOf("testShouldFlushMeta"));
986     desc.addFamily(new HColumnDescriptor("foo".getBytes()));
987     HRegion r =
988         HRegion.createHRegion(hri, testDir, conf, desc,
989             wFactory.getWAL(hri.getEncodedNameAsBytes(), hri.getTable().getNamespace()));
990     HRegion.addRegionToMETA(meta, r);
991     edge.setCurrentTimeMillis(1234 + 100);
992     StringBuffer sb = new StringBuffer();
993     assertTrue(meta.shouldFlush(sb) == false);
994     edge.setCurrentTimeMillis(edge.currentTime() + HRegion.SYSTEM_CACHE_FLUSH_INTERVAL + 1);
995     assertTrue(meta.shouldFlush(sb) == true);
996   }
997 
998   private class EnvironmentEdgeForMemstoreTest implements EnvironmentEdge {
999     long t = 1234;
1000     @Override
1001     public long currentTime() {
1002       return t; 
1003     }
1004     public void setCurrentTimeMillis(long t) {
1005       this.t = t;
1006     }
1007   }
1008 
1009   /**
1010    * Adds {@link #ROW_COUNT} rows and {@link #QUALIFIER_COUNT}
1011    * @param hmc Instance to add rows to.
1012    * @return How many rows we added.
1013    * @throws IOException
1014    */
1015   private int addRows(final MemStore hmc) {
1016     return addRows(hmc, HConstants.LATEST_TIMESTAMP);
1017   }
1018 
1019   /**
1020    * Adds {@link #ROW_COUNT} rows and {@link #QUALIFIER_COUNT}
1021    * @param hmc Instance to add rows to.
1022    * @return How many rows we added.
1023    * @throws IOException
1024    */
1025   private int addRows(final MemStore hmc, final long ts) {
1026     for (int i = 0; i < ROW_COUNT; i++) {
1027       long timestamp = ts == HConstants.LATEST_TIMESTAMP?
1028         System.currentTimeMillis(): ts;
1029       for (int ii = 0; ii < QUALIFIER_COUNT; ii++) {
1030         byte [] row = Bytes.toBytes(i);
1031         byte [] qf = makeQualifier(i, ii);
1032         hmc.add(new KeyValue(row, FAMILY, qf, timestamp, qf));
1033       }
1034     }
1035     return ROW_COUNT;
1036   }
1037 
1038   private long runSnapshot(final DefaultMemStore hmc) throws UnexpectedStateException {
1039     // Save off old state.
1040     int oldHistorySize = hmc.snapshotSection.getCellSkipListSet().sizeForTests();
1041     MemStoreSnapshot snapshot = hmc.snapshot();
1042     // Make some assertions about what just happened.
1043     assertTrue("History size has not increased", oldHistorySize <
1044       hmc.snapshotSection.getCellSkipListSet().sizeForTests());
1045     long t = memstore.timeOfOldestEdit();
1046     assertTrue("Time of oldest edit is not Long.MAX_VALUE", t == Long.MAX_VALUE);
1047     hmc.clearSnapshot(snapshot.getId());
1048     return t;
1049   }
1050 
1051   private void isExpectedRowWithoutTimestamps(final int rowIndex,
1052       List<Cell> kvs) {
1053     int i = 0;
1054     for (Cell kv: kvs) {
1055       byte[] expectedColname = makeQualifier(rowIndex, i++);
1056       assertTrue("Column name", CellUtil.matchingQualifier(kv, expectedColname));
1057       // Value is column name as bytes.  Usually result is
1058       // 100 bytes in size at least. This is the default size
1059       // for BytesWriteable.  For comparison, convert bytes to
1060       // String and trim to remove trailing null bytes.
1061       assertTrue("Content", CellUtil.matchingValue(kv, expectedColname));
1062     }
1063   }
1064 
1065   private static void addRows(int count, final MemStore mem) {
1066     long nanos = System.nanoTime();
1067 
1068     for (int i = 0 ; i < count ; i++) {
1069       if (i % 1000 == 0) {
1070 
1071         System.out.println(i + " Took for 1k usec: " + (System.nanoTime() - nanos)/1000);
1072         nanos = System.nanoTime();
1073       }
1074       long timestamp = System.currentTimeMillis();
1075 
1076       for (int ii = 0; ii < QUALIFIER_COUNT ; ii++) {
1077         byte [] row = Bytes.toBytes(i);
1078         byte [] qf = makeQualifier(i, ii);
1079         mem.add(new KeyValue(row, FAMILY, qf, timestamp, qf));
1080       }
1081     }
1082   }
1083 
1084 
1085   static void doScan(MemStore ms, int iteration) throws IOException {
1086     long nanos = System.nanoTime();
1087     KeyValueScanner s = ms.getScanners(0).get(0);
1088     s.seek(KeyValueUtil.createFirstOnRow(new byte[]{}));
1089 
1090     System.out.println(iteration + " create/seek took: " + (System.nanoTime() - nanos)/1000);
1091     int cnt=0;
1092     while(s.next() != null) ++cnt;
1093 
1094     System.out.println(iteration + " took usec: " + (System.nanoTime() - nanos) / 1000 + " for: "
1095         + cnt);
1096 
1097   }
1098 
1099   public static void main(String [] args) throws IOException {
1100     MemStore ms = new DefaultMemStore();
1101 
1102     long n1 = System.nanoTime();
1103     addRows(25000, ms);
1104     System.out.println("Took for insert: " + (System.nanoTime()-n1)/1000);
1105 
1106     System.out.println("foo");
1107 
1108     for (int i = 0 ; i < 50 ; i++)
1109       doScan(ms, i);
1110   }
1111 }
1112