001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.regionserver.compactions;
019
020import static org.apache.hadoop.hbase.regionserver.StripeStoreFileManager.OPEN_KEY;
021import static org.junit.Assert.assertEquals;
022import static org.junit.Assert.assertFalse;
023import static org.junit.Assert.assertNull;
024import static org.junit.Assert.assertTrue;
025import static org.mockito.AdditionalMatchers.aryEq;
026import static org.mockito.Matchers.any;
027import static org.mockito.Matchers.anyBoolean;
028import static org.mockito.Matchers.anyInt;
029import static org.mockito.Matchers.anyLong;
030import static org.mockito.Matchers.argThat;
031import static org.mockito.Matchers.eq;
032import static org.mockito.Matchers.isNull;
033import static org.mockito.Mockito.mock;
034import static org.mockito.Mockito.only;
035import static org.mockito.Mockito.times;
036import static org.mockito.Mockito.verify;
037import static org.mockito.Mockito.when;
038
039import java.io.IOException;
040import java.util.ArrayList;
041import java.util.Arrays;
042import java.util.Collection;
043import java.util.Iterator;
044import java.util.List;
045import java.util.OptionalLong;
046import org.apache.hadoop.conf.Configuration;
047import org.apache.hadoop.fs.Path;
048import org.apache.hadoop.hbase.Cell;
049import org.apache.hadoop.hbase.CellComparatorImpl;
050import org.apache.hadoop.hbase.HBaseClassTestRule;
051import org.apache.hadoop.hbase.HBaseConfiguration;
052import org.apache.hadoop.hbase.HColumnDescriptor;
053import org.apache.hadoop.hbase.HRegionInfo;
054import org.apache.hadoop.hbase.KeyValue;
055import org.apache.hadoop.hbase.io.hfile.HFile;
056import org.apache.hadoop.hbase.regionserver.BloomType;
057import org.apache.hadoop.hbase.regionserver.HStore;
058import org.apache.hadoop.hbase.regionserver.HStoreFile;
059import org.apache.hadoop.hbase.regionserver.InternalScanner;
060import org.apache.hadoop.hbase.regionserver.ScanInfo;
061import org.apache.hadoop.hbase.regionserver.ScanType;
062import org.apache.hadoop.hbase.regionserver.ScannerContext;
063import org.apache.hadoop.hbase.regionserver.StoreConfigInformation;
064import org.apache.hadoop.hbase.regionserver.StoreFileReader;
065import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
066import org.apache.hadoop.hbase.regionserver.StripeMultiFileWriter;
067import org.apache.hadoop.hbase.regionserver.StripeStoreConfig;
068import org.apache.hadoop.hbase.regionserver.StripeStoreFileManager;
069import org.apache.hadoop.hbase.regionserver.StripeStoreFlusher;
070import org.apache.hadoop.hbase.regionserver.compactions.StripeCompactionPolicy.StripeInformationProvider;
071import org.apache.hadoop.hbase.regionserver.compactions.TestCompactor.StoreFileWritersCapture;
072import org.apache.hadoop.hbase.regionserver.throttle.NoLimitThroughputController;
073import org.apache.hadoop.hbase.testclassification.RegionServerTests;
074import org.apache.hadoop.hbase.testclassification.SmallTests;
075import org.apache.hadoop.hbase.util.Bytes;
076import org.apache.hadoop.hbase.util.ConcatenatedLists;
077import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
078import org.apache.hadoop.hbase.util.ManualEnvironmentEdge;
079import org.junit.ClassRule;
080import org.junit.Test;
081import org.junit.experimental.categories.Category;
082import org.junit.runner.RunWith;
083import org.junit.runners.Parameterized;
084import org.junit.runners.Parameterized.Parameter;
085import org.junit.runners.Parameterized.Parameters;
086import org.mockito.ArgumentMatcher;
087
088import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableList;
089import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
090
091@RunWith(Parameterized.class)
092@Category({RegionServerTests.class, SmallTests.class})
093public class TestStripeCompactionPolicy {
094
095  @ClassRule
096  public static final HBaseClassTestRule CLASS_RULE =
097      HBaseClassTestRule.forClass(TestStripeCompactionPolicy.class);
098
099  private static final byte[] KEY_A = Bytes.toBytes("aaa");
100  private static final byte[] KEY_B = Bytes.toBytes("bbb");
101  private static final byte[] KEY_C = Bytes.toBytes("ccc");
102  private static final byte[] KEY_D = Bytes.toBytes("ddd");
103  private static final byte[] KEY_E = Bytes.toBytes("eee");
104  private static final KeyValue KV_A = new KeyValue(KEY_A, 0L);
105  private static final KeyValue KV_B = new KeyValue(KEY_B, 0L);
106  private static final KeyValue KV_C = new KeyValue(KEY_C, 0L);
107  private static final KeyValue KV_D = new KeyValue(KEY_D, 0L);
108  private static final KeyValue KV_E = new KeyValue(KEY_E, 0L);
109
110
111  private static long defaultSplitSize = 18;
112  private static float defaultSplitCount = 1.8F;
113  private final static int defaultInitialCount = 1;
114  private static long defaultTtl = 1000 * 1000;
115
116  @Parameters(name = "{index}: usePrivateReaders={0}")
117  public static Iterable<Object[]> data() {
118    return Arrays.asList(new Object[] { true }, new Object[] { false });
119  }
120
121  @Parameter
122  public boolean usePrivateReaders;
123  @Test
124  public void testNoStripesFromFlush() throws Exception {
125    Configuration conf = HBaseConfiguration.create();
126    conf.setBoolean(StripeStoreConfig.FLUSH_TO_L0_KEY, true);
127    StripeCompactionPolicy policy = createPolicy(conf);
128    StripeInformationProvider si = createStripesL0Only(0, 0);
129
130    KeyValue[] input = new KeyValue[] { KV_A, KV_B, KV_C, KV_D, KV_E };
131    KeyValue[][] expected = new KeyValue[][] { input };
132    verifyFlush(policy, si, input, expected, null);
133  }
134
135  @Test
136  public void testOldStripesFromFlush() throws Exception {
137    StripeCompactionPolicy policy = createPolicy(HBaseConfiguration.create());
138    StripeInformationProvider si = createStripes(0, KEY_C, KEY_D);
139
140    KeyValue[] input = new KeyValue[] { KV_B, KV_C, KV_C, KV_D, KV_E };
141    KeyValue[][] expected = new KeyValue[][] { new KeyValue[] { KV_B },
142        new KeyValue[] { KV_C, KV_C }, new KeyValue[] {  KV_D, KV_E } };
143    verifyFlush(policy, si, input, expected, new byte[][] { OPEN_KEY, KEY_C, KEY_D, OPEN_KEY });
144  }
145
146  @Test
147  public void testNewStripesFromFlush() throws Exception {
148    StripeCompactionPolicy policy = createPolicy(HBaseConfiguration.create());
149    StripeInformationProvider si = createStripesL0Only(0, 0);
150    KeyValue[] input = new KeyValue[] { KV_B, KV_C, KV_C, KV_D, KV_E };
151    // Starts with one stripe; unlike flush results, must have metadata
152    KeyValue[][] expected = new KeyValue[][] { input };
153    verifyFlush(policy, si, input, expected, new byte[][] { OPEN_KEY, OPEN_KEY });
154  }
155
156  @Test
157  public void testSingleStripeCompaction() throws Exception {
158    // Create a special policy that only compacts single stripes, using standard methods.
159    Configuration conf = HBaseConfiguration.create();
160    // Test depends on this not being set to pass.  Default breaks test.  TODO: Revisit.
161    conf.unset("hbase.hstore.compaction.min.size");
162    conf.setFloat(CompactionConfiguration.HBASE_HSTORE_COMPACTION_RATIO_KEY, 1.0F);
163    conf.setInt(StripeStoreConfig.MIN_FILES_KEY, 3);
164    conf.setInt(StripeStoreConfig.MAX_FILES_KEY, 4);
165    conf.setLong(StripeStoreConfig.SIZE_TO_SPLIT_KEY, 1000); // make sure the are no splits
166    StoreConfigInformation sci = mock(StoreConfigInformation.class);
167    StripeStoreConfig ssc = new StripeStoreConfig(conf, sci);
168    StripeCompactionPolicy policy = new StripeCompactionPolicy(conf, sci, ssc) {
169      @Override
170      public StripeCompactionRequest selectCompaction(StripeInformationProvider si,
171          List<HStoreFile> filesCompacting, boolean isOffpeak) throws IOException {
172        if (!filesCompacting.isEmpty()) return null;
173        return selectSingleStripeCompaction(si, false, false, isOffpeak);
174      }
175
176      @Override
177      public boolean needsCompactions(
178          StripeInformationProvider si, List<HStoreFile> filesCompacting) {
179        if (!filesCompacting.isEmpty()) return false;
180        return needsSingleStripeCompaction(si);
181      }
182    };
183
184    // No compaction due to min files or ratio
185    StripeInformationProvider si = createStripesWithSizes(0, 0,
186        new Long[] { 2L }, new Long[] { 3L, 3L }, new Long[] { 5L, 1L });
187    verifyNoCompaction(policy, si);
188    // No compaction due to min files or ratio - will report needed, but not do any.
189    si = createStripesWithSizes(0, 0,
190        new Long[] { 2L }, new Long[] { 3L, 3L }, new Long[] { 5L, 1L, 1L });
191    assertNull(policy.selectCompaction(si, al(), false));
192    assertTrue(policy.needsCompactions(si, al()));
193    // One stripe has possible compaction
194    si = createStripesWithSizes(0, 0,
195        new Long[] { 2L }, new Long[] { 3L, 3L }, new Long[] { 5L, 4L, 3L });
196    verifySingleStripeCompaction(policy, si, 2, null);
197    // Several stripes have possible compactions; choose best quality (removes most files)
198    si = createStripesWithSizes(0, 0,
199        new Long[] { 3L, 2L, 2L }, new Long[] { 2L, 2L, 1L }, new Long[] { 3L, 2L, 2L, 1L });
200    verifySingleStripeCompaction(policy, si, 2, null);
201    si = createStripesWithSizes(0, 0,
202        new Long[] { 5L }, new Long[] { 3L, 2L, 2L, 1L }, new Long[] { 3L, 2L, 2L });
203    verifySingleStripeCompaction(policy, si, 1, null);
204    // Or with smallest files, if the count is the same
205    si = createStripesWithSizes(0, 0,
206        new Long[] { 3L, 3L, 3L }, new Long[] { 3L, 1L, 2L }, new Long[] { 3L, 2L, 2L });
207    verifySingleStripeCompaction(policy, si, 1, null);
208    // Verify max count is respected.
209    si = createStripesWithSizes(0, 0, new Long[] { 5L }, new Long[] { 5L, 4L, 4L, 4L, 4L });
210    List<HStoreFile> sfs = si.getStripes().get(1).subList(1, 5);
211    verifyCompaction(policy, si, sfs, null, 1, null, si.getStartRow(1), si.getEndRow(1), true);
212    // Verify ratio is applied.
213    si = createStripesWithSizes(0, 0, new Long[] { 5L }, new Long[] { 50L, 4L, 4L, 4L, 4L });
214    sfs = si.getStripes().get(1).subList(1, 5);
215    verifyCompaction(policy, si, sfs, null, 1, null, si.getStartRow(1), si.getEndRow(1), true);
216  }
217
218  @Test
219  public void testWithParallelCompaction() throws Exception {
220    // TODO: currently only one compaction at a time per store is allowed. If this changes,
221    //       the appropriate file exclusion testing would need to be done in respective tests.
222    assertNull(createPolicy(HBaseConfiguration.create()).selectCompaction(
223        mock(StripeInformationProvider.class), al(createFile()), false));
224  }
225
226  @Test
227  public void testWithReferences() throws Exception {
228    StripeCompactionPolicy policy = createPolicy(HBaseConfiguration.create());
229    StripeCompactor sc = mock(StripeCompactor.class);
230    HStoreFile ref = createFile();
231    when(ref.isReference()).thenReturn(true);
232    StripeInformationProvider si = mock(StripeInformationProvider.class);
233    Collection<HStoreFile> sfs = al(ref, createFile());
234    when(si.getStorefiles()).thenReturn(sfs);
235
236    assertTrue(policy.needsCompactions(si, al()));
237    StripeCompactionPolicy.StripeCompactionRequest scr = policy.selectCompaction(si, al(), false);
238    // UnmodifiableCollection does not implement equals so we need to change it here to a
239    // collection that implements it.
240    assertEquals(si.getStorefiles(), new ArrayList<>(scr.getRequest().getFiles()));
241    scr.execute(sc, NoLimitThroughputController.INSTANCE, null);
242    verify(sc, only()).compact(eq(scr.getRequest()), anyInt(), anyLong(), aryEq(OPEN_KEY),
243      aryEq(OPEN_KEY), aryEq(OPEN_KEY), aryEq(OPEN_KEY),
244      any(), any());
245  }
246
247  @Test
248  public void testInitialCountFromL0() throws Exception {
249    Configuration conf = HBaseConfiguration.create();
250    conf.setInt(StripeStoreConfig.MIN_FILES_L0_KEY, 2);
251    StripeCompactionPolicy policy = createPolicy(
252        conf, defaultSplitSize, defaultSplitCount, 2, false);
253    StripeCompactionPolicy.StripeInformationProvider si = createStripesL0Only(3, 8);
254    verifyCompaction(policy, si, si.getStorefiles(), true, 2, 12L, OPEN_KEY, OPEN_KEY, true);
255    si = createStripesL0Only(3, 10); // If result would be too large, split into smaller parts.
256    verifyCompaction(policy, si, si.getStorefiles(), true, 3, 10L, OPEN_KEY, OPEN_KEY, true);
257    policy = createPolicy(conf, defaultSplitSize, defaultSplitCount, 6, false);
258    verifyCompaction(policy, si, si.getStorefiles(), true, 6, 5L, OPEN_KEY, OPEN_KEY, true);
259  }
260
261  @Test
262  public void testExistingStripesFromL0() throws Exception {
263    Configuration conf = HBaseConfiguration.create();
264    conf.setInt(StripeStoreConfig.MIN_FILES_L0_KEY, 3);
265    StripeCompactionPolicy.StripeInformationProvider si = createStripes(3, KEY_A);
266    verifyCompaction(
267        createPolicy(conf), si, si.getLevel0Files(), null, null, si.getStripeBoundaries());
268  }
269
270  @Test
271  public void testNothingToCompactFromL0() throws Exception {
272    Configuration conf = HBaseConfiguration.create();
273    conf.setInt(StripeStoreConfig.MIN_FILES_L0_KEY, 4);
274    StripeCompactionPolicy.StripeInformationProvider si = createStripesL0Only(3, 10);
275    StripeCompactionPolicy policy = createPolicy(conf);
276    verifyNoCompaction(policy, si);
277
278    si = createStripes(3, KEY_A);
279    verifyNoCompaction(policy, si);
280  }
281
282  @Test
283  public void testSplitOffStripe() throws Exception {
284    Configuration conf = HBaseConfiguration.create();
285    // Test depends on this not being set to pass.  Default breaks test.  TODO: Revisit.
286    conf.unset("hbase.hstore.compaction.min.size");
287    // First test everything with default split count of 2, then split into more.
288    conf.setInt(StripeStoreConfig.MIN_FILES_KEY, 2);
289    Long[] toSplit = new Long[] { defaultSplitSize - 2, 1L, 1L };
290    Long[] noSplit = new Long[] { defaultSplitSize - 2, 1L };
291    long splitTargetSize = (long)(defaultSplitSize / defaultSplitCount);
292    // Don't split if not eligible for compaction.
293    StripeCompactionPolicy.StripeInformationProvider si =
294        createStripesWithSizes(0, 0, new Long[] { defaultSplitSize - 2, 2L });
295    assertNull(createPolicy(conf).selectCompaction(si, al(), false));
296    // Make sure everything is eligible.
297    conf.setFloat(CompactionConfiguration.HBASE_HSTORE_COMPACTION_RATIO_KEY, 500f);
298    StripeCompactionPolicy policy = createPolicy(conf);
299    verifyWholeStripesCompaction(policy, si, 0, 0, null, 2, splitTargetSize);
300    // Add some extra stripes...
301    si = createStripesWithSizes(0, 0, noSplit, noSplit, toSplit);
302    verifyWholeStripesCompaction(policy, si, 2, 2, null, 2, splitTargetSize);
303    // In the middle.
304    si = createStripesWithSizes(0, 0, noSplit, toSplit, noSplit);
305    verifyWholeStripesCompaction(policy, si, 1, 1, null, 2, splitTargetSize);
306    // No split-off with different config (larger split size).
307    // However, in this case some eligible stripe will just be compacted alone.
308    StripeCompactionPolicy specPolicy = createPolicy(
309        conf, defaultSplitSize + 1, defaultSplitCount, defaultInitialCount, false);
310    verifySingleStripeCompaction(specPolicy, si, 1, null);
311  }
312
313  @Test
314  public void testSplitOffStripeOffPeak() throws Exception {
315    // for HBASE-11439
316    Configuration conf = HBaseConfiguration.create();
317
318    // Test depends on this not being set to pass.  Default breaks test.  TODO: Revisit.
319    conf.unset("hbase.hstore.compaction.min.size");
320
321    conf.setInt(StripeStoreConfig.MIN_FILES_KEY, 2);
322    // Select the last 2 files.
323    StripeCompactionPolicy.StripeInformationProvider si =
324        createStripesWithSizes(0, 0, new Long[] { defaultSplitSize - 2, 1L, 1L });
325    assertEquals(2, createPolicy(conf).selectCompaction(si, al(), false).getRequest().getFiles()
326        .size());
327    // Make sure everything is eligible in offpeak.
328    conf.setFloat("hbase.hstore.compaction.ratio.offpeak", 500f);
329    assertEquals(3, createPolicy(conf).selectCompaction(si, al(), true).getRequest().getFiles()
330        .size());
331  }
332
333  @Test
334  public void testSplitOffStripeDropDeletes() throws Exception {
335    Configuration conf = HBaseConfiguration.create();
336    conf.setInt(StripeStoreConfig.MIN_FILES_KEY, 2);
337    StripeCompactionPolicy policy = createPolicy(conf);
338    Long[] toSplit = new Long[] { defaultSplitSize / 2, defaultSplitSize / 2 };
339    Long[] noSplit = new Long[] { 1L };
340    long splitTargetSize = (long)(defaultSplitSize / defaultSplitCount);
341
342    // Verify the deletes can be dropped if there are no L0 files.
343    StripeCompactionPolicy.StripeInformationProvider si =
344        createStripesWithSizes(0, 0, noSplit, toSplit);
345    verifyWholeStripesCompaction(policy, si, 1, 1,    true, null, splitTargetSize);
346    // But cannot be dropped if there are.
347    si = createStripesWithSizes(2, 2, noSplit, toSplit);
348    verifyWholeStripesCompaction(policy, si, 1, 1,    false, null, splitTargetSize);
349  }
350
351  @SuppressWarnings("unchecked")
352  @Test
353  public void testMergeExpiredFiles() throws Exception {
354    ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
355    long now = defaultTtl + 2;
356    edge.setValue(now);
357    EnvironmentEdgeManager.injectEdge(edge);
358    try {
359      HStoreFile expiredFile = createFile(), notExpiredFile = createFile();
360      when(expiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl - 1);
361      when(notExpiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl + 1);
362      List<HStoreFile> expired = Lists.newArrayList(expiredFile, expiredFile);
363      List<HStoreFile> notExpired = Lists.newArrayList(notExpiredFile, notExpiredFile);
364      List<HStoreFile> mixed = Lists.newArrayList(expiredFile, notExpiredFile);
365
366      StripeCompactionPolicy policy = createPolicy(HBaseConfiguration.create(),
367          defaultSplitSize, defaultSplitCount, defaultInitialCount, true);
368      // Merge expired if there are eligible stripes.
369      StripeCompactionPolicy.StripeInformationProvider si =
370          createStripesWithFiles(expired, expired, expired);
371      verifyWholeStripesCompaction(policy, si, 0, 2, null, 1, Long.MAX_VALUE, false);
372      // Don't merge if nothing expired.
373      si = createStripesWithFiles(notExpired, notExpired, notExpired);
374      assertNull(policy.selectCompaction(si, al(), false));
375      // Merge one expired stripe with next.
376      si = createStripesWithFiles(notExpired, expired, notExpired);
377      verifyWholeStripesCompaction(policy, si, 1, 2, null, 1, Long.MAX_VALUE, false);
378      // Merge the biggest run out of multiple options.
379      // Merge one expired stripe with next.
380      si = createStripesWithFiles(notExpired, expired, notExpired, expired, expired, notExpired);
381      verifyWholeStripesCompaction(policy, si, 3, 4, null, 1, Long.MAX_VALUE, false);
382      // Stripe with a subset of expired files is not merged.
383      si = createStripesWithFiles(expired, expired, notExpired, expired, mixed);
384      verifyWholeStripesCompaction(policy, si, 0, 1, null, 1, Long.MAX_VALUE, false);
385    } finally {
386      EnvironmentEdgeManager.reset();
387    }
388  }
389
390  @SuppressWarnings("unchecked")
391  @Test
392  public void testMergeExpiredStripes() throws Exception {
393    // HBASE-11397
394    ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
395    long now = defaultTtl + 2;
396    edge.setValue(now);
397    EnvironmentEdgeManager.injectEdge(edge);
398    try {
399      HStoreFile expiredFile = createFile(), notExpiredFile = createFile();
400      when(expiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl - 1);
401      when(notExpiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl + 1);
402      List<HStoreFile> expired = Lists.newArrayList(expiredFile, expiredFile);
403      List<HStoreFile> notExpired = Lists.newArrayList(notExpiredFile, notExpiredFile);
404
405      StripeCompactionPolicy policy =
406          createPolicy(HBaseConfiguration.create(), defaultSplitSize, defaultSplitCount,
407            defaultInitialCount, true);
408
409      // Merge all three expired stripes into one.
410      StripeCompactionPolicy.StripeInformationProvider si =
411          createStripesWithFiles(expired, expired, expired);
412      verifyMergeCompatcion(policy, si, 0, 2);
413
414      // Merge two adjacent expired stripes into one.
415      si = createStripesWithFiles(notExpired, expired, notExpired, expired, expired, notExpired);
416      verifyMergeCompatcion(policy, si, 3, 4);
417    } finally {
418      EnvironmentEdgeManager.reset();
419    }
420  }
421
422  @SuppressWarnings("unchecked")
423  private static StripeCompactionPolicy.StripeInformationProvider createStripesWithFiles(
424      List<HStoreFile>... stripeFiles) throws Exception {
425    return createStripesWithFiles(createBoundaries(stripeFiles.length),
426        Lists.newArrayList(stripeFiles), new ArrayList<>());
427  }
428
429  @Test
430  public void testSingleStripeDropDeletes() throws Exception {
431    Configuration conf = HBaseConfiguration.create();
432    // Test depends on this not being set to pass.  Default breaks test.  TODO: Revisit.
433    conf.unset("hbase.hstore.compaction.min.size");
434    StripeCompactionPolicy policy = createPolicy(conf);
435    // Verify the deletes can be dropped if there are no L0 files.
436    Long[][] stripes = new Long[][] { new Long[] { 3L, 2L, 2L, 2L }, new Long[] { 6L } };
437    StripeInformationProvider si = createStripesWithSizes(0, 0, stripes);
438    verifySingleStripeCompaction(policy, si, 0, true);
439    // But cannot be dropped if there are.
440    si = createStripesWithSizes(2, 2, stripes);
441    verifySingleStripeCompaction(policy, si, 0, false);
442    // Unless there are enough to cause L0 compaction.
443    si = createStripesWithSizes(6, 2, stripes);
444    ConcatenatedLists<HStoreFile> sfs = new ConcatenatedLists<>();
445    sfs.addSublist(si.getLevel0Files());
446    sfs.addSublist(si.getStripes().get(0));
447    verifyCompaction(
448        policy, si, sfs, si.getStartRow(0), si.getEndRow(0), si.getStripeBoundaries());
449    // If we cannot actually compact all files in some stripe, L0 is chosen.
450    si = createStripesWithSizes(6, 2,
451        new Long[][] { new Long[] { 10L, 1L, 1L, 1L, 1L }, new Long[] { 12L } });
452    verifyCompaction(policy, si, si.getLevel0Files(), null, null, si.getStripeBoundaries());
453    // even if L0 has no file
454    // if all files of stripe aren't selected, delete must not be dropped.
455    stripes = new Long[][] { new Long[] { 100L, 3L, 2L, 2L, 2L }, new Long[] { 6L } };
456    si = createStripesWithSizes(0, 0, stripes);
457    List<HStoreFile> compactFile = new ArrayList<>();
458    Iterator<HStoreFile> iter = si.getStripes().get(0).listIterator(1);
459    while (iter.hasNext()) {
460        compactFile.add(iter.next());
461    }
462    verifyCompaction(policy, si, compactFile, false, 1, null, si.getStartRow(0), si.getEndRow(0),
463      true);
464  }
465
466  /********* HELPER METHODS ************/
467  private static StripeCompactionPolicy createPolicy(
468      Configuration conf) throws Exception {
469    return createPolicy(conf, defaultSplitSize, defaultSplitCount, defaultInitialCount, false);
470  }
471
472  private static StripeCompactionPolicy createPolicy(Configuration conf,
473      long splitSize, float splitCount, int initialCount, boolean hasTtl) throws Exception {
474    conf.setLong(StripeStoreConfig.SIZE_TO_SPLIT_KEY, splitSize);
475    conf.setFloat(StripeStoreConfig.SPLIT_PARTS_KEY, splitCount);
476    conf.setInt(StripeStoreConfig.INITIAL_STRIPE_COUNT_KEY, initialCount);
477    StoreConfigInformation sci = mock(StoreConfigInformation.class);
478    when(sci.getStoreFileTtl()).thenReturn(hasTtl ? defaultTtl : Long.MAX_VALUE);
479    StripeStoreConfig ssc = new StripeStoreConfig(conf, sci);
480    return new StripeCompactionPolicy(conf, sci, ssc);
481  }
482
483  private static ArrayList<HStoreFile> al(HStoreFile... sfs) {
484    return new ArrayList<>(Arrays.asList(sfs));
485  }
486
487  private void verifyMergeCompatcion(StripeCompactionPolicy policy, StripeInformationProvider si,
488      int from, int to) throws Exception {
489    StripeCompactionPolicy.StripeCompactionRequest scr = policy.selectCompaction(si, al(), false);
490    Collection<HStoreFile> sfs = getAllFiles(si, from, to);
491    verifyCollectionsEqual(sfs, scr.getRequest().getFiles());
492
493    // All the Stripes are expired, so the Compactor will not create any Writers. We need to create
494    // an empty file to preserve metadata
495    StripeCompactor sc = createCompactor();
496    List<Path> paths = scr.execute(sc, NoLimitThroughputController.INSTANCE, null);
497    assertEquals(1, paths.size());
498  }
499
500  /**
501   * Verify the compaction that includes several entire stripes.
502   * @param policy Policy to test.
503   * @param si Stripe information pre-set with stripes to test.
504   * @param from Starting stripe.
505   * @param to Ending stripe (inclusive).
506   * @param dropDeletes Whether to drop deletes from compaction range.
507   * @param count Expected # of resulting stripes, null if not checked.
508   * @param size Expected target stripe size, null if not checked.
509   */
510  private void verifyWholeStripesCompaction(StripeCompactionPolicy policy,
511      StripeInformationProvider si, int from, int to, Boolean dropDeletes,
512      Integer count, Long size, boolean needsCompaction) throws IOException {
513    verifyCompaction(policy, si, getAllFiles(si, from, to), dropDeletes,
514        count, size, si.getStartRow(from), si.getEndRow(to), needsCompaction);
515  }
516
517  private void verifyWholeStripesCompaction(StripeCompactionPolicy policy,
518      StripeInformationProvider si, int from, int to, Boolean dropDeletes,
519      Integer count, Long size) throws IOException {
520    verifyWholeStripesCompaction(policy, si, from, to, dropDeletes, count, size, true);
521  }
522
523  private void verifySingleStripeCompaction(StripeCompactionPolicy policy,
524      StripeInformationProvider si, int index, Boolean dropDeletes) throws IOException {
525    verifyWholeStripesCompaction(policy, si, index, index, dropDeletes, 1, null, true);
526  }
527
528  /**
529   * Verify no compaction is needed or selected.
530   * @param policy Policy to test.
531   * @param si Stripe information pre-set with stripes to test.
532   */
533  private void verifyNoCompaction(
534      StripeCompactionPolicy policy, StripeInformationProvider si) throws IOException {
535    assertNull(policy.selectCompaction(si, al(), false));
536    assertFalse(policy.needsCompactions(si, al()));
537  }
538
539  /**
540   * Verify arbitrary compaction.
541   * @param policy Policy to test.
542   * @param si Stripe information pre-set with stripes to test.
543   * @param sfs Files that should be compacted.
544   * @param dropDeletesFrom Row from which to drop deletes.
545   * @param dropDeletesTo Row to which to drop deletes.
546   * @param boundaries Expected target stripe boundaries.
547   */
548  private void verifyCompaction(StripeCompactionPolicy policy, StripeInformationProvider si,
549      Collection<HStoreFile> sfs, byte[] dropDeletesFrom, byte[] dropDeletesTo,
550      final List<byte[]> boundaries) throws Exception {
551    StripeCompactor sc = mock(StripeCompactor.class);
552    assertTrue(policy.needsCompactions(si, al()));
553    StripeCompactionPolicy.StripeCompactionRequest scr = policy.selectCompaction(si, al(), false);
554    verifyCollectionsEqual(sfs, scr.getRequest().getFiles());
555    scr.execute(sc, NoLimitThroughputController.INSTANCE, null);
556    verify(sc, times(1)).compact(eq(scr.getRequest()), argThat(new ArgumentMatcher<List<byte[]>>() {
557      @Override
558      public boolean matches(List<byte[]> argument) {
559        List<byte[]> other = argument;
560        if (other.size() != boundaries.size()) return false;
561        for (int i = 0; i < other.size(); ++i) {
562          if (!Bytes.equals(other.get(i), boundaries.get(i))) return false;
563        }
564        return true;
565      }
566    }), dropDeletesFrom == null ? isNull(byte[].class) : aryEq(dropDeletesFrom),
567      dropDeletesTo == null ? isNull(byte[].class) : aryEq(dropDeletesTo),
568      any(), any());
569  }
570
571  /**
572   * Verify arbitrary compaction.
573   * @param policy Policy to test.
574   * @param si Stripe information pre-set with stripes to test.
575   * @param sfs Files that should be compacted.
576   * @param dropDeletes Whether to drop deletes from compaction range.
577   * @param count Expected # of resulting stripes, null if not checked.
578   * @param size Expected target stripe size, null if not checked.
579   * @param start Left boundary of the compaction.
580   * @param end Right boundary of the compaction.
581   */
582  private void verifyCompaction(StripeCompactionPolicy policy, StripeInformationProvider si,
583      Collection<HStoreFile> sfs, Boolean dropDeletes, Integer count, Long size,
584      byte[] start, byte[] end, boolean needsCompaction) throws IOException {
585    StripeCompactor sc = mock(StripeCompactor.class);
586    assertTrue(!needsCompaction || policy.needsCompactions(si, al()));
587    StripeCompactionPolicy.StripeCompactionRequest scr = policy.selectCompaction(si, al(), false);
588    verifyCollectionsEqual(sfs, scr.getRequest().getFiles());
589    scr.execute(sc, NoLimitThroughputController.INSTANCE, null);
590    verify(sc, times(1)).compact(eq(scr.getRequest()),
591      count == null ? anyInt() : eq(count.intValue()),
592      size == null ? anyLong() : eq(size.longValue()), aryEq(start), aryEq(end),
593      dropDeletesMatcher(dropDeletes, start), dropDeletesMatcher(dropDeletes, end),
594      any(), any());
595  }
596
597  /** Verify arbitrary flush. */
598  protected void verifyFlush(StripeCompactionPolicy policy, StripeInformationProvider si,
599      KeyValue[] input, KeyValue[][] expected, byte[][] boundaries) throws IOException {
600    StoreFileWritersCapture writers = new StoreFileWritersCapture();
601    StripeStoreFlusher.StripeFlushRequest req = policy.selectFlush(CellComparatorImpl.COMPARATOR, si,
602      input.length);
603    StripeMultiFileWriter mw = req.createWriter();
604    mw.init(null, writers);
605    for (KeyValue kv : input) {
606      mw.append(kv);
607    }
608    boolean hasMetadata = boundaries != null;
609    mw.commitWriters(0, false);
610    writers.verifyKvs(expected, true, hasMetadata);
611    if (hasMetadata) {
612      writers.verifyBoundaries(boundaries);
613    }
614  }
615
616
617  private byte[] dropDeletesMatcher(Boolean dropDeletes, byte[] value) {
618    return dropDeletes == null ? any()
619            : (dropDeletes.booleanValue() ? aryEq(value) : isNull(byte[].class));
620  }
621
622  private void verifyCollectionsEqual(Collection<HStoreFile> sfs, Collection<HStoreFile> scr) {
623    // Dumb.
624    assertEquals(sfs.size(), scr.size());
625    assertTrue(scr.containsAll(sfs));
626  }
627
628  private static List<HStoreFile> getAllFiles(
629      StripeInformationProvider si, int fromStripe, int toStripe) {
630    ArrayList<HStoreFile> expected = new ArrayList<>();
631    for (int i = fromStripe; i <= toStripe; ++i) {
632      expected.addAll(si.getStripes().get(i));
633    }
634    return expected;
635  }
636
637  /**
638   * @param l0Count Number of L0 files.
639   * @param boundaries Target boundaries.
640   * @return Mock stripes.
641   */
642  private static StripeInformationProvider createStripes(
643      int l0Count, byte[]... boundaries) throws Exception {
644    List<Long> l0Sizes = new ArrayList<>();
645    for (int i = 0; i < l0Count; ++i) {
646      l0Sizes.add(5L);
647    }
648    List<List<Long>> sizes = new ArrayList<>();
649    for (int i = 0; i <= boundaries.length; ++i) {
650      sizes.add(Arrays.asList(Long.valueOf(5)));
651    }
652    return createStripes(Arrays.asList(boundaries), sizes, l0Sizes);
653  }
654
655  /**
656   * @param l0Count Number of L0 files.
657   * @param l0Size Size of each file.
658   * @return Mock stripes.
659   */
660  private static StripeInformationProvider createStripesL0Only(
661      int l0Count, long l0Size) throws Exception {
662    List<Long> l0Sizes = new ArrayList<>();
663    for (int i = 0; i < l0Count; ++i) {
664      l0Sizes.add(l0Size);
665    }
666    return createStripes(null, new ArrayList<>(), l0Sizes);
667  }
668
669  /**
670   * @param l0Count Number of L0 files.
671   * @param l0Size Size of each file.
672   * @param sizes Sizes of the files; each sub-array representing a stripe.
673   * @return Mock stripes.
674   */
675  private static StripeInformationProvider createStripesWithSizes(
676      int l0Count, long l0Size, Long[]... sizes) throws Exception {
677    ArrayList<List<Long>> sizeList = new ArrayList<>(sizes.length);
678    for (Long[] size : sizes) {
679      sizeList.add(Arrays.asList(size));
680    }
681    return createStripesWithSizes(l0Count, l0Size, sizeList);
682  }
683
684  private static StripeInformationProvider createStripesWithSizes(
685      int l0Count, long l0Size, List<List<Long>> sizes) throws Exception {
686    List<byte[]> boundaries = createBoundaries(sizes.size());
687    List<Long> l0Sizes = new ArrayList<>();
688    for (int i = 0; i < l0Count; ++i) {
689      l0Sizes.add(l0Size);
690    }
691    return createStripes(boundaries, sizes, l0Sizes);
692  }
693
694  private static List<byte[]> createBoundaries(int stripeCount) {
695    byte[][] keys = new byte[][] { KEY_A, KEY_B, KEY_C, KEY_D, KEY_E };
696    assert stripeCount <= keys.length + 1;
697    List<byte[]> boundaries = new ArrayList<>();
698    boundaries.addAll(Arrays.asList(keys).subList(0, stripeCount - 1));
699    return boundaries;
700  }
701
702  private static StripeInformationProvider createStripes(List<byte[]> boundaries,
703      List<List<Long>> stripeSizes, List<Long> l0Sizes) throws Exception {
704    List<List<HStoreFile>> stripeFiles = new ArrayList<>(stripeSizes.size());
705    for (List<Long> sizes : stripeSizes) {
706      List<HStoreFile> sfs = new ArrayList<>(sizes.size());
707      for (Long size : sizes) {
708        sfs.add(createFile(size));
709      }
710      stripeFiles.add(sfs);
711    }
712    List<HStoreFile> l0Files = new ArrayList<>();
713    for (Long size : l0Sizes) {
714      l0Files.add(createFile(size));
715    }
716    return createStripesWithFiles(boundaries, stripeFiles, l0Files);
717  }
718
719  /**
720   * This method actually does all the work.
721   */
722  private static StripeInformationProvider createStripesWithFiles(List<byte[]> boundaries,
723      List<List<HStoreFile>> stripeFiles, List<HStoreFile> l0Files) throws Exception {
724    ArrayList<ImmutableList<HStoreFile>> stripes = new ArrayList<>();
725    ArrayList<byte[]> boundariesList = new ArrayList<>();
726    StripeInformationProvider si = mock(StripeInformationProvider.class);
727    if (!stripeFiles.isEmpty()) {
728      assert stripeFiles.size() == (boundaries.size() + 1);
729      boundariesList.add(OPEN_KEY);
730      for (int i = 0; i <= boundaries.size(); ++i) {
731        byte[] startKey = ((i == 0) ? OPEN_KEY : boundaries.get(i - 1));
732        byte[] endKey = ((i == boundaries.size()) ? OPEN_KEY : boundaries.get(i));
733        boundariesList.add(endKey);
734        for (HStoreFile sf : stripeFiles.get(i)) {
735          setFileStripe(sf, startKey, endKey);
736        }
737        stripes.add(ImmutableList.copyOf(stripeFiles.get(i)));
738        when(si.getStartRow(eq(i))).thenReturn(startKey);
739        when(si.getEndRow(eq(i))).thenReturn(endKey);
740      }
741    }
742    ConcatenatedLists<HStoreFile> sfs = new ConcatenatedLists<>();
743    sfs.addAllSublists(stripes);
744    sfs.addSublist(l0Files);
745    when(si.getStorefiles()).thenReturn(sfs);
746    when(si.getStripes()).thenReturn(stripes);
747    when(si.getStripeBoundaries()).thenReturn(boundariesList);
748    when(si.getStripeCount()).thenReturn(stripes.size());
749    when(si.getLevel0Files()).thenReturn(l0Files);
750    return si;
751  }
752
753  private static HStoreFile createFile(long size) throws Exception {
754    HStoreFile sf = mock(HStoreFile.class);
755    when(sf.getPath()).thenReturn(new Path("moo"));
756    StoreFileReader r = mock(StoreFileReader.class);
757    when(r.getEntries()).thenReturn(size);
758    when(r.length()).thenReturn(size);
759    when(r.getBloomFilterType()).thenReturn(BloomType.NONE);
760    when(r.getHFileReader()).thenReturn(mock(HFile.Reader.class));
761    when(r.getStoreFileScanner(anyBoolean(), anyBoolean(), anyBoolean(), anyLong(), anyLong(),
762      anyBoolean())).thenReturn(mock(StoreFileScanner.class));
763    when(sf.getReader()).thenReturn(r);
764    when(sf.getBulkLoadTimestamp()).thenReturn(OptionalLong.empty());
765    return sf;
766  }
767
768  private static HStoreFile createFile() throws Exception {
769    return createFile(0);
770  }
771
772  private static void setFileStripe(HStoreFile sf, byte[] startKey, byte[] endKey) {
773    when(sf.getMetadataValue(StripeStoreFileManager.STRIPE_START_KEY)).thenReturn(startKey);
774    when(sf.getMetadataValue(StripeStoreFileManager.STRIPE_END_KEY)).thenReturn(endKey);
775  }
776
777  private StripeCompactor createCompactor() throws Exception {
778    HColumnDescriptor col = new HColumnDescriptor(Bytes.toBytes("foo"));
779    StoreFileWritersCapture writers = new StoreFileWritersCapture();
780    HStore store = mock(HStore.class);
781    HRegionInfo info = mock(HRegionInfo.class);
782    when(info.getRegionNameAsString()).thenReturn("testRegion");
783    when(store.getColumnFamilyDescriptor()).thenReturn(col);
784    when(store.getRegionInfo()).thenReturn(info);
785    when(
786      store.createWriterInTmp(anyLong(), any(), anyBoolean(),
787        anyBoolean(), anyBoolean(), anyBoolean())).thenAnswer(writers);
788
789    Configuration conf = HBaseConfiguration.create();
790    conf.setBoolean("hbase.regionserver.compaction.private.readers", usePrivateReaders);
791    final Scanner scanner = new Scanner();
792    return new StripeCompactor(conf, store) {
793      @Override
794      protected InternalScanner createScanner(HStore store, ScanInfo scanInfo,
795          List<StoreFileScanner> scanners, long smallestReadPoint, long earliestPutTs,
796          byte[] dropDeletesFromRow, byte[] dropDeletesToRow) throws IOException {
797        return scanner;
798      }
799
800      @Override
801      protected InternalScanner createScanner(HStore store, ScanInfo scanInfo,
802          List<StoreFileScanner> scanners, ScanType scanType, long smallestReadPoint,
803          long earliestPutTs) throws IOException {
804        return scanner;
805      }
806    };
807  }
808
809  private static class Scanner implements InternalScanner {
810    private final ArrayList<KeyValue> kvs;
811
812    public Scanner(KeyValue... kvs) {
813      this.kvs = new ArrayList<>(Arrays.asList(kvs));
814    }
815
816    @Override
817    public boolean next(List<Cell> result, ScannerContext scannerContext)
818        throws IOException {
819      if (kvs.isEmpty()) return false;
820      result.add(kvs.remove(0));
821      return !kvs.isEmpty();
822    }
823
824    @Override
825    public void close() throws IOException {
826    }
827  }
828}