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 static org.apache.hadoop.hbase.regionserver.StripeStoreFileManager.OPEN_KEY;
22  import static org.junit.Assert.assertEquals;
23  import static org.junit.Assert.assertFalse;
24  import static org.junit.Assert.assertNull;
25  import static org.junit.Assert.assertTrue;
26  import static org.junit.Assert.fail;
27  
28  import java.io.IOException;
29  import java.util.ArrayList;
30  import java.util.Arrays;
31  import java.util.Collection;
32  import java.util.Collections;
33  import java.util.Iterator;
34  import java.util.List;
35  
36  import org.apache.hadoop.conf.Configuration;
37  import org.apache.hadoop.fs.FileSystem;
38  import org.apache.hadoop.fs.Path;
39  import org.apache.hadoop.hbase.HBaseConfiguration;
40  import org.apache.hadoop.hbase.HBaseTestingUtility;
41  import org.apache.hadoop.hbase.HConstants;
42  import org.apache.hadoop.hbase.KeyValue;
43  import org.apache.hadoop.hbase.KeyValue.KVComparator;
44  import org.apache.hadoop.hbase.SmallTests;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.junit.After;
47  import org.junit.Before;
48  import org.junit.Test;
49  import org.junit.experimental.categories.Category;
50  import org.mockito.Mockito;
51  
52  
53  @Category(SmallTests.class)
54  public class TestStripeStoreFileManager {
55    private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
56    private static final Path BASEDIR =
57        TEST_UTIL.getDataTestDir(TestStripeStoreFileManager.class.getSimpleName());
58    private static final Path CFDIR = HStore.getStoreHomedir(BASEDIR, "region", Bytes.toBytes("cf"));
59  
60    private static final byte[] KEY_A = Bytes.toBytes("aaa");
61    private static final byte[] KEY_B = Bytes.toBytes("bbb");
62    private static final byte[] KEY_C = Bytes.toBytes("ccc");
63    private static final byte[] KEY_D = Bytes.toBytes("ddd");
64  
65    private static final KeyValue KV_A = new KeyValue(KEY_A, 0L);
66    private static final KeyValue KV_B = new KeyValue(KEY_B, 0L);
67    private static final KeyValue KV_C = new KeyValue(KEY_C, 0L);
68    private static final KeyValue KV_D = new KeyValue(KEY_D, 0L);
69  
70    @Before
71    public void setUp() throws Exception {
72      FileSystem fs = TEST_UTIL.getTestFileSystem();
73      if (!fs.mkdirs(CFDIR)) {
74        throw new IOException("Cannot create test directory " + CFDIR);
75      }
76    }
77  
78    @After
79    public void tearDown() throws Exception {
80      FileSystem fs = TEST_UTIL.getTestFileSystem();
81      if (fs.exists(CFDIR) && !fs.delete(CFDIR, true)) {
82        throw new IOException("Cannot delete test directory " + CFDIR);
83      }
84    }
85  
86    @Test
87    public void testInsertFilesIntoL0() throws Exception {
88      StripeStoreFileManager manager = createManager();
89      MockStoreFile sf = createFile();
90      manager.insertNewFiles(al(sf));
91      assertEquals(1, manager.getStorefileCount());
92      Collection<StoreFile> filesForGet = manager.getFilesForScanOrGet(true, KEY_A, KEY_A);
93      assertEquals(1, filesForGet.size());
94      assertTrue(filesForGet.contains(sf));
95  
96      // Add some stripes and make sure we get this file for every stripe.
97      manager.addCompactionResults(al(), al(createFile(OPEN_KEY, KEY_B),
98          createFile(KEY_B, OPEN_KEY)));
99      assertTrue(manager.getFilesForScanOrGet(true, KEY_A, KEY_A).contains(sf));
100     assertTrue(manager.getFilesForScanOrGet(true, KEY_C, KEY_C).contains(sf));
101   }
102 
103   @Test
104   public void testClearFiles() throws Exception {
105     StripeStoreFileManager manager = createManager();
106     manager.insertNewFiles(al(createFile()));
107     manager.insertNewFiles(al(createFile()));
108     manager.addCompactionResults(al(), al(createFile(OPEN_KEY, KEY_B),
109         createFile(KEY_B, OPEN_KEY)));
110     assertEquals(4, manager.getStorefileCount());
111     Collection<StoreFile> allFiles = manager.clearFiles();
112     assertEquals(4, allFiles.size());
113     assertEquals(0, manager.getStorefileCount());
114     assertEquals(0, manager.getStorefiles().size());
115   }
116 
117   private static ArrayList<StoreFile> dumpIterator(Iterator<StoreFile> iter) {
118     ArrayList<StoreFile> result = new ArrayList<StoreFile>();
119     for (; iter.hasNext(); result.add(iter.next()));
120     return result;
121   }
122 
123   @Test
124   public void testRowKeyBefore() throws Exception {
125     StripeStoreFileManager manager = createManager();
126     StoreFile l0File = createFile(), l0File2 = createFile();
127     manager.insertNewFiles(al(l0File));
128     manager.insertNewFiles(al(l0File2));
129     // Get candidate files.
130     Iterator<StoreFile> sfs = manager.getCandidateFilesForRowKeyBefore(KV_B);
131     sfs.next();
132     sfs.remove();
133     // Suppose we found a candidate in this file... make sure L0 file remaining is not removed.
134     sfs = manager.updateCandidateFilesForRowKeyBefore(sfs, KV_B, KV_A);
135     assertTrue(sfs.hasNext());
136     // Now add some stripes (remove L0 file too)
137     MockStoreFile stripe0a = createFile(0, 100, OPEN_KEY, KEY_B),
138         stripe1 = createFile(KEY_B, OPEN_KEY);
139     manager.addCompactionResults(al(l0File), al(stripe0a, stripe1));
140     // If we want a key <= KEY_A, we should get everything except stripe1.
141     ArrayList<StoreFile> sfsDump = dumpIterator(manager.getCandidateFilesForRowKeyBefore(KV_A));
142     assertEquals(2, sfsDump.size());
143     assertTrue(sfsDump.contains(stripe0a));
144     assertFalse(sfsDump.contains(stripe1));
145     // If we want a key <= KEY_B, we should get everything since lower bound is inclusive.
146     sfsDump = dumpIterator(manager.getCandidateFilesForRowKeyBefore(KV_B));
147     assertEquals(3, sfsDump.size());
148     assertTrue(sfsDump.contains(stripe1));
149     // For KEY_D, we should also get everything.
150     sfsDump = dumpIterator(manager.getCandidateFilesForRowKeyBefore(KV_D));
151     assertEquals(3, sfsDump.size());
152     // Suppose in the first file we found candidate with KEY_C.
153     // Then, stripe0 no longer matters and should be removed, but stripe1 should stay.
154     sfs = manager.getCandidateFilesForRowKeyBefore(KV_D);
155     sfs.next(); // Skip L0 file.
156     sfs.remove();
157     sfs = manager.updateCandidateFilesForRowKeyBefore(sfs, KV_D, KV_C);
158     assertEquals(stripe1, sfs.next());
159     assertFalse(sfs.hasNext());
160     // Add one more, later, file to stripe0, remove the last annoying L0 file.
161     // This file should be returned in preference to older L0 file; also, after we get
162     // a candidate from the first file, the old one should not be removed.
163     StoreFile stripe0b = createFile(0, 101, OPEN_KEY, KEY_B);
164     manager.addCompactionResults(al(l0File2), al(stripe0b));
165     sfs = manager.getCandidateFilesForRowKeyBefore(KV_A);
166     assertEquals(stripe0b, sfs.next());
167     sfs.remove();
168     sfs = manager.updateCandidateFilesForRowKeyBefore(sfs, KV_A, KV_A);
169     assertEquals(stripe0a, sfs.next());
170   }
171 
172   @Test
173   public void testGetSplitPointEdgeCases() throws Exception {
174     StripeStoreFileManager manager = createManager();
175     // No files => no split.
176     assertNull(manager.getSplitPoint());
177 
178     // If there are no stripes, should pick midpoint from the biggest file in L0.
179     MockStoreFile sf5 = createFile(5, 0);
180     sf5.splitPoint = new byte[1];
181     manager.insertNewFiles(al(sf5));
182     manager.insertNewFiles(al(createFile(1, 0)));
183     assertEquals(sf5.splitPoint, manager.getSplitPoint());
184 
185     // Same if there's one stripe but the biggest file is still in L0.
186     manager.addCompactionResults(al(), al(createFile(2, 0, OPEN_KEY, OPEN_KEY)));
187     assertEquals(sf5.splitPoint, manager.getSplitPoint());
188 
189     // If the biggest file is in the stripe, should get from it.
190     MockStoreFile sf6 = createFile(6, 0, OPEN_KEY, OPEN_KEY);
191     sf6.splitPoint = new byte[1];
192     manager.addCompactionResults(al(), al(sf6));
193     assertEquals(sf6.splitPoint, manager.getSplitPoint());
194   }
195 
196   @Test
197   public void testGetStripeBoundarySplits() throws Exception {
198     /* First number - split must be after this stripe; further numbers - stripes */
199     verifySplitPointScenario(5, false, 0f,     2, 1, 1, 1, 1, 1, 10);
200     verifySplitPointScenario(0, false, 0f,     6, 3, 1, 1, 2);
201     verifySplitPointScenario(2, false, 0f,     1, 1, 1, 1, 2);
202     verifySplitPointScenario(0, false, 0f,     5, 4);
203     verifySplitPointScenario(2, false, 0f,     5, 2, 5, 5, 5);
204   }
205 
206   @Test
207   public void testGetUnbalancedSplits() throws Exception {
208     /* First number - split must be inside/after this stripe; further numbers - stripes */
209     verifySplitPointScenario(0, false, 2.1f,      4, 4, 4); // 8/4 is less than 2.1f
210     verifySplitPointScenario(1, true,  1.5f,      4, 4, 4); // 8/4 > 6/6
211     verifySplitPointScenario(1, false, 1.1f,      3, 4, 1, 1, 2, 2); // 7/6 < 8/5
212     verifySplitPointScenario(1, false, 1.1f,      3, 6, 1, 1, 2, 2); // 9/6 == 9/6
213     verifySplitPointScenario(1, true,  1.1f,      3, 8, 1, 1, 2, 2); // 11/6 > 10/7
214     verifySplitPointScenario(3, false, 1.1f,      2, 2, 1, 1, 4, 3); // reverse order
215     verifySplitPointScenario(4, true,  1.1f,      2, 2, 1, 1, 8, 3); // reverse order
216     verifySplitPointScenario(0, true,  1.5f,      10, 4); // 10/4 > 9/5
217     verifySplitPointScenario(0, false, 1.4f,      6, 4);  // 6/4 == 6/4
218     verifySplitPointScenario(1, true,  1.5f,      4, 10); // reverse just in case
219     verifySplitPointScenario(0, false, 1.4f,      4, 6);  // reverse just in case
220   }
221 
222 
223   /**
224    * Verifies scenario for finding a split point.
225    * @param splitPointAfter Stripe to expect the split point at/after.
226    * @param shouldSplitStripe If true, the split point is expected in the middle of the above
227    *                          stripe; if false, should be at the end.
228    * @param splitRatioToVerify Maximum split imbalance ratio.
229    * @param sizes Stripe sizes.
230    */
231   private void verifySplitPointScenario(int splitPointAfter, boolean shouldSplitStripe,
232       float splitRatioToVerify, int... sizes) throws Exception {
233     assertTrue(sizes.length > 1);
234     ArrayList<StoreFile> sfs = new ArrayList<StoreFile>();
235     for (int sizeIx = 0; sizeIx < sizes.length; ++sizeIx) {
236       byte[] startKey = (sizeIx == 0) ? OPEN_KEY : Bytes.toBytes(sizeIx - 1);
237       byte[] endKey = (sizeIx == sizes.length - 1) ? OPEN_KEY : Bytes.toBytes(sizeIx);
238       MockStoreFile sf = createFile(sizes[sizeIx], 0, startKey, endKey);
239       sf.splitPoint = Bytes.toBytes(-sizeIx); // set split point to the negative index
240       sfs.add(sf);
241     }
242 
243     Configuration conf = HBaseConfiguration.create();
244     if (splitRatioToVerify != 0) {
245       conf.setFloat(StripeStoreConfig.MAX_REGION_SPLIT_IMBALANCE_KEY, splitRatioToVerify);
246     }
247     StripeStoreFileManager manager = createManager(al(), conf);
248     manager.addCompactionResults(al(), sfs);
249     int result = Bytes.toInt(manager.getSplitPoint());
250     // Either end key and thus positive index, or "middle" of the file and thus negative index.
251     assertEquals(splitPointAfter * (shouldSplitStripe ? -1 : 1), result);
252   }
253 
254   private static byte[] keyAfter(byte[] key) {
255     return Arrays.copyOf(key, key.length + 1);
256   }
257 
258   @Test
259   public void testGetFilesForGetAndScan() throws Exception {
260     StripeStoreFileManager manager = createManager();
261     verifyGetAndScanScenario(manager, null, null);
262     verifyGetAndScanScenario(manager, KEY_B, KEY_C);
263 
264     // Populate one L0 file.
265     MockStoreFile sf0 = createFile();
266     manager.insertNewFiles(al(sf0));
267     verifyGetAndScanScenario(manager, null, null,   sf0);
268     verifyGetAndScanScenario(manager, null, KEY_C,  sf0);
269     verifyGetAndScanScenario(manager, KEY_B, null,  sf0);
270     verifyGetAndScanScenario(manager, KEY_B, KEY_C, sf0);
271 
272     // Populate a bunch of files for stripes, keep L0.
273     MockStoreFile sfA = createFile(OPEN_KEY, KEY_A);
274     MockStoreFile sfB = createFile(KEY_A, KEY_B);
275     MockStoreFile sfC = createFile(KEY_B, KEY_C);
276     MockStoreFile sfD = createFile(KEY_C, KEY_D);
277     MockStoreFile sfE = createFile(KEY_D, OPEN_KEY);
278     manager.addCompactionResults(al(), al(sfA, sfB, sfC, sfD, sfE));
279 
280     verifyGetAndScanScenario(manager, null, null,              sf0, sfA, sfB, sfC, sfD, sfE);
281     verifyGetAndScanScenario(manager, keyAfter(KEY_A), null,   sf0, sfB, sfC, sfD, sfE);
282     verifyGetAndScanScenario(manager, null, keyAfter(KEY_C),   sf0, sfA, sfB, sfC, sfD);
283     verifyGetAndScanScenario(manager, KEY_B, null,             sf0, sfC, sfD, sfE);
284     verifyGetAndScanScenario(manager, null, KEY_C,             sf0, sfA, sfB, sfC, sfD);
285     verifyGetAndScanScenario(manager, KEY_B, keyAfter(KEY_B),  sf0, sfC);
286     verifyGetAndScanScenario(manager, keyAfter(KEY_A), KEY_B,  sf0, sfB, sfC);
287     verifyGetAndScanScenario(manager, KEY_D, KEY_D,            sf0, sfE);
288     verifyGetAndScanScenario(manager, keyAfter(KEY_B), keyAfter(KEY_C), sf0, sfC, sfD);
289   }
290 
291   private void verifyGetAndScanScenario(StripeStoreFileManager manager,
292       byte[] start, byte[] end, StoreFile... results) throws Exception {
293     verifyGetOrScanScenario(manager, true, start, end, results);
294     verifyGetOrScanScenario(manager, false, start, end, results);
295   }
296 
297   @Test
298   @SuppressWarnings("unchecked")
299   public void testLoadFilesWithRecoverableBadFiles() throws Exception {
300     // In L0, there will be file w/o metadata (real L0, 3 files with invalid metadata, and 3
301     // files that overlap valid stripes in various ways). Note that the 4th way to overlap the
302     // stripes will cause the structure to be mostly scraped, and is tested separately.
303     ArrayList<StoreFile> validStripeFiles = al(createFile(OPEN_KEY, KEY_B),
304         createFile(KEY_B, KEY_C), createFile(KEY_C, OPEN_KEY),
305         createFile(KEY_C, OPEN_KEY));
306     ArrayList<StoreFile> filesToGoToL0 = al(createFile(), createFile(null, KEY_A),
307         createFile(KEY_D, null), createFile(KEY_D, KEY_A), createFile(keyAfter(KEY_A), KEY_C),
308         createFile(OPEN_KEY, KEY_D), createFile(KEY_D, keyAfter(KEY_D)));
309     ArrayList<StoreFile> allFilesToGo = flattenLists(validStripeFiles, filesToGoToL0);
310     Collections.shuffle(allFilesToGo);
311     StripeStoreFileManager manager = createManager(allFilesToGo);
312     List<StoreFile> l0Files = manager.getLevel0Files();
313     assertEquals(filesToGoToL0.size(), l0Files.size());
314     for (StoreFile sf : filesToGoToL0) {
315       assertTrue(l0Files.contains(sf));
316     }
317     verifyAllFiles(manager, allFilesToGo);
318   }
319 
320   @Test
321   public void testLoadFilesWithBadStripe() throws Exception {
322     // Current "algorithm" will see the after-B key before C key, add it as valid stripe,
323     // and then fail all other stripes. So everything would end up in L0.
324     ArrayList<StoreFile> allFilesToGo = al(createFile(OPEN_KEY, KEY_B),
325         createFile(KEY_B, KEY_C), createFile(KEY_C, OPEN_KEY),
326         createFile(KEY_B, keyAfter(KEY_B)));
327     Collections.shuffle(allFilesToGo);
328     StripeStoreFileManager manager = createManager(allFilesToGo);
329     assertEquals(allFilesToGo.size(), manager.getLevel0Files().size());
330   }
331 
332   @Test
333   public void testLoadFilesWithGaps() throws Exception {
334     // Stripes must not have gaps. If they do, everything goes to L0.
335     StripeStoreFileManager manager =
336       createManager(al(createFile(OPEN_KEY, KEY_B), createFile(KEY_C, OPEN_KEY)));
337     assertEquals(2, manager.getLevel0Files().size());
338     // Just one open stripe should be ok.
339     manager = createManager(al(createFile(OPEN_KEY, OPEN_KEY)));
340     assertEquals(0, manager.getLevel0Files().size());
341     assertEquals(1, manager.getStorefileCount());
342   }
343 
344   @Test
345   public void testLoadFilesAfterSplit() throws Exception {
346     // If stripes are good but have non-open ends, they must be treated as open ends.
347     MockStoreFile sf = createFile(KEY_B, KEY_C);
348     StripeStoreFileManager manager = createManager(al(createFile(OPEN_KEY, KEY_B), sf));
349     assertEquals(0, manager.getLevel0Files().size());
350     // Here, [B, C] is logically [B, inf), so we should be able to compact it to that only.
351     verifyInvalidCompactionScenario(manager, al(sf), al(createFile(KEY_B, KEY_C)));
352     manager.addCompactionResults(al(sf), al(createFile(KEY_B, OPEN_KEY)));
353     // Do the same for other variants.
354     manager = createManager(al(sf, createFile(KEY_C, OPEN_KEY)));
355     verifyInvalidCompactionScenario(manager, al(sf), al(createFile(KEY_B, KEY_C)));
356     manager.addCompactionResults(al(sf), al(createFile(OPEN_KEY, KEY_C)));
357     manager = createManager(al(sf));
358     verifyInvalidCompactionScenario(manager, al(sf), al(createFile(KEY_B, KEY_C)));
359     manager.addCompactionResults(al(sf), al(createFile(OPEN_KEY, OPEN_KEY)));
360   }
361 
362   @Test
363   public void testAddingCompactionResults() throws Exception {
364     StripeStoreFileManager manager = createManager();
365     // First, add some L0 files and "compact" one with new stripe creation.
366     StoreFile sf_L0_0a = createFile(), sf_L0_0b = createFile();
367     manager.insertNewFiles(al(sf_L0_0a, sf_L0_0b));
368 
369     // Try compacting with invalid new branches (gaps, overlaps) - no effect.
370     verifyInvalidCompactionScenario(manager, al(sf_L0_0a), al(createFile(OPEN_KEY, KEY_B)));
371     verifyInvalidCompactionScenario(manager, al(sf_L0_0a), al(createFile(OPEN_KEY, KEY_B),
372         createFile(KEY_C, OPEN_KEY)));
373     verifyInvalidCompactionScenario(manager, al(sf_L0_0a), al(createFile(OPEN_KEY, KEY_B),
374         createFile(KEY_B, OPEN_KEY), createFile(KEY_A, KEY_D)));
375     verifyInvalidCompactionScenario(manager, al(sf_L0_0a), al(createFile(OPEN_KEY, KEY_B),
376         createFile(KEY_A, KEY_B), createFile(KEY_B, OPEN_KEY)));
377 
378     StoreFile sf_i2B_0 = createFile(OPEN_KEY, KEY_B);
379     StoreFile sf_B2C_0 = createFile(KEY_B, KEY_C);
380     StoreFile sf_C2i_0 = createFile(KEY_C, OPEN_KEY);
381     manager.addCompactionResults(al(sf_L0_0a), al(sf_i2B_0, sf_B2C_0, sf_C2i_0));
382     verifyAllFiles(manager, al(sf_L0_0b, sf_i2B_0, sf_B2C_0, sf_C2i_0));
383 
384     // Add another l0 file, "compact" both L0 into two stripes
385     StoreFile sf_L0_1 = createFile();
386     StoreFile sf_i2B_1 = createFile(OPEN_KEY, KEY_B);
387     StoreFile sf_B2C_1 = createFile(KEY_B, KEY_C);
388     manager.insertNewFiles(al(sf_L0_1));
389     manager.addCompactionResults(al(sf_L0_0b, sf_L0_1), al(sf_i2B_1, sf_B2C_1));
390     verifyAllFiles(manager, al(sf_i2B_0, sf_B2C_0, sf_C2i_0, sf_i2B_1, sf_B2C_1));
391 
392     // Try compacting with invalid file (no metadata) - should add files to L0.
393     StoreFile sf_L0_2 = createFile(null, null);
394     manager.addCompactionResults(al(), al(sf_L0_2));
395     verifyAllFiles(manager, al(sf_i2B_0, sf_B2C_0, sf_C2i_0, sf_i2B_1, sf_B2C_1, sf_L0_2));
396     // Remove it...
397     manager.addCompactionResults(al(sf_L0_2), al());
398 
399     // Do regular compaction in the first stripe.
400     StoreFile sf_i2B_3 = createFile(OPEN_KEY, KEY_B);
401     manager.addCompactionResults(al(sf_i2B_0, sf_i2B_1), al(sf_i2B_3));
402     verifyAllFiles(manager, al(sf_B2C_0, sf_C2i_0, sf_B2C_1, sf_i2B_3));
403 
404     // Rebalance two stripes.
405     StoreFile sf_B2D_4 = createFile(KEY_B, KEY_D);
406     StoreFile sf_D2i_4 = createFile(KEY_D, OPEN_KEY);
407     manager.addCompactionResults(al(sf_B2C_0, sf_C2i_0, sf_B2C_1), al(sf_B2D_4, sf_D2i_4));
408     verifyAllFiles(manager, al(sf_i2B_3, sf_B2D_4, sf_D2i_4));
409 
410     // Split the first stripe.
411     StoreFile sf_i2A_5 = createFile(OPEN_KEY, KEY_A);
412     StoreFile sf_A2B_5 = createFile(KEY_A, KEY_B);
413     manager.addCompactionResults(al(sf_i2B_3), al(sf_i2A_5, sf_A2B_5));
414     verifyAllFiles(manager, al(sf_B2D_4, sf_D2i_4, sf_i2A_5, sf_A2B_5));
415 
416     // Split the middle stripe.
417     StoreFile sf_B2C_6 = createFile(KEY_B, KEY_C);
418     StoreFile sf_C2D_6 = createFile(KEY_C, KEY_D);
419     manager.addCompactionResults(al(sf_B2D_4), al(sf_B2C_6, sf_C2D_6));
420     verifyAllFiles(manager, al(sf_D2i_4, sf_i2A_5, sf_A2B_5, sf_B2C_6, sf_C2D_6));
421 
422     // Merge two different middle stripes.
423     StoreFile sf_A2C_7 = createFile(KEY_A, KEY_C);
424     manager.addCompactionResults(al(sf_A2B_5, sf_B2C_6), al(sf_A2C_7));
425     verifyAllFiles(manager, al(sf_D2i_4, sf_i2A_5, sf_C2D_6, sf_A2C_7));
426 
427     // Merge lower half.
428     StoreFile sf_i2C_8 = createFile(OPEN_KEY, KEY_C);
429     manager.addCompactionResults(al(sf_i2A_5, sf_A2C_7), al(sf_i2C_8));
430     verifyAllFiles(manager, al(sf_D2i_4, sf_C2D_6, sf_i2C_8));
431 
432     // Merge all.
433     StoreFile sf_i2i_9 = createFile(OPEN_KEY, OPEN_KEY);
434     manager.addCompactionResults(al(sf_D2i_4, sf_C2D_6, sf_i2C_8), al(sf_i2i_9));
435     verifyAllFiles(manager, al(sf_i2i_9));
436   }
437 
438   @Test
439   public void testCompactionAndFlushConflict() throws Exception {
440     // Add file flush into stripes
441     StripeStoreFileManager sfm = createManager();
442     assertEquals(0, sfm.getStripeCount());
443     StoreFile sf_i2c = createFile(OPEN_KEY, KEY_C), sf_c2i = createFile(KEY_C, OPEN_KEY);
444     sfm.insertNewFiles(al(sf_i2c, sf_c2i));
445     assertEquals(2, sfm.getStripeCount());
446     // Now try to add conflicting flush - should throw.
447     StoreFile sf_i2d = createFile(OPEN_KEY, KEY_D), sf_d2i = createFile(KEY_D, OPEN_KEY);
448     sfm.insertNewFiles(al(sf_i2d, sf_d2i));
449     assertEquals(2, sfm.getStripeCount());
450     assertEquals(2, sfm.getLevel0Files().size());
451     verifyGetAndScanScenario(sfm, KEY_C, KEY_C, sf_i2d, sf_d2i, sf_c2i);
452     // Remove these files.
453     sfm.addCompactionResults(al(sf_i2d, sf_d2i), al());
454     assertEquals(0, sfm.getLevel0Files().size());
455     // Add another file to stripe; then "rebalance" stripes w/o it - the file, which was
456     // presumably flushed during compaction, should go to L0.
457     StoreFile sf_i2c_2 = createFile(OPEN_KEY, KEY_C);
458     sfm.insertNewFiles(al(sf_i2c_2));
459     sfm.addCompactionResults(al(sf_i2c, sf_c2i), al(sf_i2d, sf_d2i));
460     assertEquals(1, sfm.getLevel0Files().size());
461     verifyGetAndScanScenario(sfm, KEY_C, KEY_C, sf_i2d, sf_i2c_2);
462   }
463 
464   @Test
465   public void testEmptyResultsForStripes() throws Exception {
466     // Test that we can compact L0 into a subset of stripes.
467     StripeStoreFileManager manager = createManager();
468     StoreFile sf0a = createFile();
469     StoreFile sf0b = createFile();
470     manager.insertNewFiles(al(sf0a));
471     manager.insertNewFiles(al(sf0b));
472     ArrayList<StoreFile> compacted = al(createFile(OPEN_KEY, KEY_B),
473         createFile(KEY_B, KEY_C), createFile(KEY_C, OPEN_KEY));
474     manager.addCompactionResults(al(sf0a), compacted);
475     // Next L0 compaction only produces file for the first and last stripe.
476     ArrayList<StoreFile> compacted2 = al(createFile(OPEN_KEY, KEY_B), createFile(KEY_C, OPEN_KEY));
477     manager.addCompactionResults(al(sf0b), compacted2);
478     compacted.addAll(compacted2);
479     verifyAllFiles(manager, compacted);
480   }
481 
482   @Test
483   public void testPriority() throws Exception {
484     // Expected priority, file limit, stripe count, files per stripe, l0 files.
485     testPriorityScenario(5,    5, 0, 0, 0);
486     testPriorityScenario(2,    5, 0, 0, 3);
487     testPriorityScenario(4,   25, 5, 1, 0); // example case.
488     testPriorityScenario(3,   25, 5, 1, 1); // L0 files counts for all stripes.
489     testPriorityScenario(3,   25, 5, 2, 0); // file to each stripe - same as one L0 file.
490     testPriorityScenario(2,   25, 5, 4, 0); // 1 is priority user, so 2 is returned.
491     testPriorityScenario(2,   25, 5, 4, 4); // don't return higher than user unless over limit.
492     testPriorityScenario(2,   25, 5, 1, 10); // same.
493     testPriorityScenario(0,   25, 5, 4, 5); // at limit.
494     testPriorityScenario(-5,  25, 5, 6, 0); // over limit!
495     testPriorityScenario(-1,  25, 0, 0, 26); // over limit with just L0
496   }
497 
498   private void testPriorityScenario(int expectedPriority,
499       int limit, int stripes, int filesInStripe, int l0Files) throws Exception
500   {
501     final byte[][] keys = { KEY_A, KEY_B, KEY_C, KEY_D };
502     assertTrue(stripes <= keys.length + 1);
503     Configuration conf = TEST_UTIL.getConfiguration();
504     conf.setInt("hbase.hstore.blockingStoreFiles", limit);
505     StripeStoreFileManager sfm = createManager(al(), conf);
506     for (int i = 0; i < l0Files; ++i) {
507       sfm.insertNewFiles(al(createFile()));
508     }
509     for (int i = 0; i < filesInStripe; ++i) {
510       ArrayList<StoreFile> stripe = new ArrayList<StoreFile>();
511       for (int j = 0; j < stripes; ++j) {
512         stripe.add(createFile(
513             (j == 0) ? OPEN_KEY : keys[j - 1], (j == stripes - 1) ? OPEN_KEY : keys[j]));
514       }
515       sfm.addCompactionResults(al(), stripe);
516     }
517     assertEquals(expectedPriority, sfm.getStoreCompactionPriority());
518   }
519 
520   private void verifyInvalidCompactionScenario(StripeStoreFileManager manager,
521       ArrayList<StoreFile> filesToCompact, ArrayList<StoreFile> filesToInsert) throws Exception {
522     Collection<StoreFile> allFiles = manager.getStorefiles();
523     try {
524        manager.addCompactionResults(filesToCompact, filesToInsert);
525        fail("Should have thrown");
526     } catch (IOException ex) {
527       // Ignore it.
528     }
529     verifyAllFiles(manager, allFiles); // must have the same files.
530   }
531 
532   private void verifyGetOrScanScenario(StripeStoreFileManager manager, boolean isGet,
533       byte[] start, byte[] end, StoreFile... results) throws Exception {
534     verifyGetOrScanScenario(manager, isGet, start, end, Arrays.asList(results));
535   }
536 
537   private void verifyGetOrScanScenario(StripeStoreFileManager manager, boolean isGet,
538       byte[] start, byte[] end, Collection<StoreFile> results) throws Exception {
539     start = start != null ? start : HConstants.EMPTY_START_ROW;
540     end = end != null ? end : HConstants.EMPTY_END_ROW;
541     Collection<StoreFile> sfs = manager.getFilesForScanOrGet(isGet, start, end);
542     assertEquals(results.size(), sfs.size());
543     for (StoreFile result : results) {
544       assertTrue(sfs.contains(result));
545     }
546   }
547 
548   private void verifyAllFiles(
549       StripeStoreFileManager manager, Collection<StoreFile> results) throws Exception {
550     verifyGetOrScanScenario(manager, false, null, null, results);
551   }
552 
553   // TODO: replace with Mockito?
554   private static MockStoreFile createFile(
555       long size, long seqNum, byte[] startKey, byte[] endKey) throws Exception {
556     FileSystem fs = TEST_UTIL.getTestFileSystem();
557     Path testFilePath = StoreFile.getUniqueFile(fs, CFDIR);
558     fs.create(testFilePath).close();
559     MockStoreFile sf = new MockStoreFile(TEST_UTIL, testFilePath, size, 0, false, seqNum);
560     if (startKey != null) {
561       sf.setMetadataValue(StripeStoreFileManager.STRIPE_START_KEY, startKey);
562     }
563     if (endKey != null) {
564       sf.setMetadataValue(StripeStoreFileManager.STRIPE_END_KEY, endKey);
565     }
566     return sf;
567   }
568 
569   private static MockStoreFile createFile(long size, long seqNum) throws Exception {
570     return createFile(size, seqNum, null, null);
571   }
572 
573   private static MockStoreFile createFile(byte[] startKey, byte[] endKey) throws Exception {
574     return createFile(0, 0, startKey, endKey);
575   }
576 
577   private static MockStoreFile createFile() throws Exception {
578     return createFile(null, null);
579   }
580 
581   private static StripeStoreFileManager createManager() throws Exception {
582     return createManager(new ArrayList<StoreFile>());
583   }
584 
585   private static StripeStoreFileManager createManager(ArrayList<StoreFile> sfs) throws Exception {
586     return createManager(sfs, TEST_UTIL.getConfiguration());
587   }
588 
589   private static StripeStoreFileManager createManager(
590       ArrayList<StoreFile> sfs, Configuration conf) throws Exception {
591     StripeStoreConfig config = new StripeStoreConfig(
592         conf, Mockito.mock(StoreConfigInformation.class));
593     StripeStoreFileManager result = new StripeStoreFileManager(new KVComparator(), conf, config);
594     result.loadFiles(sfs);
595     return result;
596   }
597 
598   private static ArrayList<StoreFile> al(StoreFile... sfs) {
599     return new ArrayList<StoreFile>(Arrays.asList(sfs));
600   }
601 
602   private static ArrayList<StoreFile> flattenLists(ArrayList<StoreFile>... sfls) {
603     ArrayList<StoreFile> result = new ArrayList<StoreFile>();
604     for (ArrayList<StoreFile> sfl : sfls) {
605       result.addAll(sfl);
606     }
607     return result;
608   }
609 }