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}