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