1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver.compactions;
20
21 import static org.apache.hadoop.hbase.regionserver.StripeStoreFileManager.OPEN_KEY;
22
23 import java.io.IOException;
24 import java.util.ArrayList;
25 import java.util.Collection;
26 import java.util.List;
27
28 import com.google.common.collect.ImmutableList;
29
30 import org.apache.commons.logging.Log;
31 import org.apache.commons.logging.LogFactory;
32 import org.apache.hadoop.conf.Configuration;
33 import org.apache.hadoop.fs.Path;
34 import org.apache.hadoop.hbase.KeyValue.KVComparator;
35 import org.apache.hadoop.hbase.classification.InterfaceAudience;
36 import org.apache.hadoop.hbase.regionserver.StoreConfigInformation;
37 import org.apache.hadoop.hbase.regionserver.StoreFile;
38 import org.apache.hadoop.hbase.regionserver.StoreUtils;
39 import org.apache.hadoop.hbase.regionserver.StripeStoreConfig;
40 import org.apache.hadoop.hbase.regionserver.StripeStoreFlusher;
41 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
42 import org.apache.hadoop.hbase.security.User;
43 import org.apache.hadoop.hbase.util.Bytes;
44 import org.apache.hadoop.hbase.util.ConcatenatedLists;
45 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
46 import org.apache.hadoop.hbase.util.Pair;
47
48
49
50
51 @InterfaceAudience.Private
52 public class StripeCompactionPolicy extends CompactionPolicy {
53 private final static Log LOG = LogFactory.getLog(StripeCompactionPolicy.class);
54
55 private ExploringCompactionPolicy stripePolicy = null;
56
57 private StripeStoreConfig config;
58
59 public StripeCompactionPolicy(
60 Configuration conf, StoreConfigInformation storeConfigInfo, StripeStoreConfig config) {
61 super(conf, storeConfigInfo);
62 this.config = config;
63 stripePolicy = new ExploringCompactionPolicy(conf, storeConfigInfo);
64 }
65
66 public List<StoreFile> preSelectFilesForCoprocessor(StripeInformationProvider si,
67 List<StoreFile> filesCompacting) {
68
69
70
71 ArrayList<StoreFile> candidateFiles = new ArrayList<StoreFile>(si.getStorefiles());
72 candidateFiles.removeAll(filesCompacting);
73 return candidateFiles;
74 }
75
76 public StripeCompactionRequest createEmptyRequest(
77 StripeInformationProvider si, CompactionRequest request) {
78
79 if (si.getStripeCount() > 0) {
80 return new BoundaryStripeCompactionRequest(request, si.getStripeBoundaries());
81 }
82 Pair<Long, Integer> targetKvsAndCount = estimateTargetKvs(
83 request.getFiles(), this.config.getInitialCount());
84 return new SplitStripeCompactionRequest(
85 request, OPEN_KEY, OPEN_KEY, targetKvsAndCount.getSecond(), targetKvsAndCount.getFirst());
86 }
87
88 public StripeStoreFlusher.StripeFlushRequest selectFlush(KVComparator comparator,
89 StripeInformationProvider si, int kvCount) {
90 if (this.config.isUsingL0Flush()) {
91
92 return new StripeStoreFlusher.StripeFlushRequest(comparator);
93 }
94 if (si.getStripeCount() == 0) {
95
96 int initialCount = this.config.getInitialCount();
97 return new StripeStoreFlusher.SizeStripeFlushRequest(comparator, initialCount,
98 kvCount / initialCount);
99 }
100
101 return new StripeStoreFlusher.BoundaryStripeFlushRequest(comparator, si.getStripeBoundaries());
102 }
103
104 public StripeCompactionRequest selectCompaction(StripeInformationProvider si,
105 List<StoreFile> filesCompacting, boolean isOffpeak) throws IOException {
106
107
108 if (!filesCompacting.isEmpty()) {
109 LOG.debug("Not selecting compaction: " + filesCompacting.size() + " files compacting");
110 return null;
111 }
112
113
114
115
116
117
118
119 Collection<StoreFile> allFiles = si.getStorefiles();
120 if (StoreUtils.hasReferences(allFiles)) {
121 LOG.debug("There are references in the store; compacting all files");
122 long targetKvs = estimateTargetKvs(allFiles, config.getInitialCount()).getFirst();
123 SplitStripeCompactionRequest request = new SplitStripeCompactionRequest(
124 allFiles, OPEN_KEY, OPEN_KEY, targetKvs);
125 request.setMajorRangeFull();
126 return request;
127 }
128
129 int stripeCount = si.getStripeCount();
130 List<StoreFile> l0Files = si.getLevel0Files();
131
132
133 boolean shouldCompactL0 = (this.config.getLevel0MinFiles() <= l0Files.size());
134 if (stripeCount == 0) {
135 if (!shouldCompactL0) return null;
136 return selectNewStripesCompaction(si);
137 }
138
139 boolean canDropDeletesNoL0 = l0Files.size() == 0;
140 if (shouldCompactL0) {
141 if (!canDropDeletesNoL0) {
142
143 StripeCompactionRequest result = selectSingleStripeCompaction(
144 si, true, canDropDeletesNoL0, isOffpeak);
145 if (result != null) return result;
146 }
147 LOG.debug("Selecting L0 compaction with " + l0Files.size() + " files");
148 return new BoundaryStripeCompactionRequest(l0Files, si.getStripeBoundaries());
149 }
150
151
152 StripeCompactionRequest result = selectExpiredMergeCompaction(si, canDropDeletesNoL0);
153 if (result != null) return result;
154
155
156
157 return selectSingleStripeCompaction(si, false, canDropDeletesNoL0, isOffpeak);
158 }
159
160 public boolean needsCompactions(StripeInformationProvider si, List<StoreFile> filesCompacting) {
161
162 return filesCompacting.isEmpty()
163 && (StoreUtils.hasReferences(si.getStorefiles())
164 || (si.getLevel0Files().size() >= this.config.getLevel0MinFiles())
165 || needsSingleStripeCompaction(si));
166 }
167
168 @Override
169 public boolean shouldPerformMajorCompaction(Collection<StoreFile> filesToCompact)
170 throws IOException {
171 return false;
172 }
173
174 @Override
175 public boolean throttleCompaction(long compactionSize) {
176 return compactionSize > comConf.getThrottlePoint();
177 }
178
179
180
181
182
183 protected boolean needsSingleStripeCompaction(StripeInformationProvider si) {
184 int minFiles = this.config.getStripeCompactMinFiles();
185 for (List<StoreFile> stripe : si.getStripes()) {
186 if (stripe.size() >= minFiles) return true;
187 }
188 return false;
189 }
190
191 protected StripeCompactionRequest selectSingleStripeCompaction(StripeInformationProvider si,
192 boolean includeL0, boolean canDropDeletesWithoutL0, boolean isOffpeak) throws IOException {
193 ArrayList<ImmutableList<StoreFile>> stripes = si.getStripes();
194
195 int bqIndex = -1;
196 List<StoreFile> bqSelection = null;
197 int stripeCount = stripes.size();
198 long bqTotalSize = -1;
199 for (int i = 0; i < stripeCount; ++i) {
200
201
202 List<StoreFile> selection = selectSimpleCompaction(stripes.get(i),
203 !canDropDeletesWithoutL0 && includeL0, isOffpeak);
204 if (selection.isEmpty()) continue;
205 long size = 0;
206 for (StoreFile sf : selection) {
207 size += sf.getReader().length();
208 }
209 if (bqSelection == null || selection.size() > bqSelection.size() ||
210 (selection.size() == bqSelection.size() && size < bqTotalSize)) {
211 bqSelection = selection;
212 bqIndex = i;
213 bqTotalSize = size;
214 }
215 }
216 if (bqSelection == null) {
217 LOG.debug("No good compaction is possible in any stripe");
218 return null;
219 }
220 List<StoreFile> filesToCompact = new ArrayList<StoreFile>(bqSelection);
221
222 int targetCount = 1;
223 long targetKvs = Long.MAX_VALUE;
224 boolean hasAllFiles = filesToCompact.size() == stripes.get(bqIndex).size();
225 String splitString = "";
226 if (hasAllFiles && bqTotalSize >= config.getSplitSize()) {
227 if (includeL0) {
228
229
230 return null;
231 }
232 Pair<Long, Integer> kvsAndCount = estimateTargetKvs(filesToCompact, config.getSplitCount());
233 targetKvs = kvsAndCount.getFirst();
234 targetCount = kvsAndCount.getSecond();
235 splitString = "; the stripe will be split into at most "
236 + targetCount + " stripes with " + targetKvs + " target KVs";
237 }
238
239 LOG.debug("Found compaction in a stripe with end key ["
240 + Bytes.toString(si.getEndRow(bqIndex)) + "], with "
241 + filesToCompact.size() + " files of total size " + bqTotalSize + splitString);
242
243
244 StripeCompactionRequest req;
245 if (includeL0) {
246 assert hasAllFiles;
247 List<StoreFile> l0Files = si.getLevel0Files();
248 LOG.debug("Adding " + l0Files.size() + " files to compaction to be able to drop deletes");
249 ConcatenatedLists<StoreFile> sfs = new ConcatenatedLists<StoreFile>();
250 sfs.addSublist(filesToCompact);
251 sfs.addSublist(l0Files);
252 req = new BoundaryStripeCompactionRequest(sfs, si.getStripeBoundaries());
253 } else {
254 req = new SplitStripeCompactionRequest(
255 filesToCompact, si.getStartRow(bqIndex), si.getEndRow(bqIndex), targetCount, targetKvs);
256 }
257 if (hasAllFiles && (canDropDeletesWithoutL0 || includeL0)) {
258 req.setMajorRange(si.getStartRow(bqIndex), si.getEndRow(bqIndex));
259 }
260 req.getRequest().setOffPeak(isOffpeak);
261 return req;
262 }
263
264
265
266
267
268
269
270 private List<StoreFile> selectSimpleCompaction(
271 List<StoreFile> sfs, boolean allFilesOnly, boolean isOffpeak) {
272 int minFilesLocal = Math.max(
273 allFilesOnly ? sfs.size() : 0, this.config.getStripeCompactMinFiles());
274 int maxFilesLocal = Math.max(this.config.getStripeCompactMaxFiles(), minFilesLocal);
275 return stripePolicy.applyCompactionPolicy(sfs, false, isOffpeak, minFilesLocal, maxFilesLocal);
276 }
277
278
279
280
281
282
283
284
285 private StripeCompactionRequest selectCompactionOfAllFiles(StripeInformationProvider si,
286 int targetStripeCount, long targetSize) {
287 Collection<StoreFile> allFiles = si.getStorefiles();
288 SplitStripeCompactionRequest request = new SplitStripeCompactionRequest(
289 allFiles, OPEN_KEY, OPEN_KEY, targetStripeCount, targetSize);
290 request.setMajorRangeFull();
291 LOG.debug("Selecting a compaction that includes all " + allFiles.size() + " files");
292 return request;
293 }
294
295 private StripeCompactionRequest selectNewStripesCompaction(StripeInformationProvider si) {
296 List<StoreFile> l0Files = si.getLevel0Files();
297 Pair<Long, Integer> kvsAndCount = estimateTargetKvs(l0Files, config.getInitialCount());
298 LOG.debug("Creating " + kvsAndCount.getSecond() + " initial stripes with "
299 + kvsAndCount.getFirst() + " kvs each via L0 compaction of " + l0Files.size() + " files");
300 SplitStripeCompactionRequest request = new SplitStripeCompactionRequest(
301 si.getLevel0Files(), OPEN_KEY, OPEN_KEY, kvsAndCount.getSecond(), kvsAndCount.getFirst());
302 request.setMajorRangeFull();
303 return request;
304 }
305
306 private StripeCompactionRequest selectExpiredMergeCompaction(
307 StripeInformationProvider si, boolean canDropDeletesNoL0) {
308 long cfTtl = this.storeConfigInfo.getStoreFileTtl();
309 if (cfTtl == Long.MAX_VALUE) {
310 return null;
311 }
312 long timestampCutoff = EnvironmentEdgeManager.currentTime() - cfTtl;
313
314 int start = -1, bestStart = -1, length = 0, bestLength = 0;
315 ArrayList<ImmutableList<StoreFile>> stripes = si.getStripes();
316 OUTER: for (int i = 0; i < stripes.size(); ++i) {
317 for (StoreFile storeFile : stripes.get(i)) {
318 if (storeFile.getReader().getMaxTimestamp() < timestampCutoff) continue;
319
320 if (length > bestLength) {
321 bestStart = start;
322 bestLength = length;
323 }
324 start = -1;
325 length = 0;
326 continue OUTER;
327 }
328 if (start == -1) {
329 start = i;
330 }
331 ++length;
332 }
333 if (length > bestLength) {
334 bestStart = start;
335 bestLength = length;
336 }
337 if (bestLength == 0) return null;
338 if (bestLength == 1) {
339
340
341
342
343 if (bestStart == (stripes.size() - 1)) return null;
344 ++bestLength;
345 }
346 LOG.debug("Merging " + bestLength + " stripes to delete expired store files");
347 int endIndex = bestStart + bestLength - 1;
348 ConcatenatedLists<StoreFile> sfs = new ConcatenatedLists<StoreFile>();
349 sfs.addAllSublists(stripes.subList(bestStart, endIndex + 1));
350 SplitStripeCompactionRequest result = new SplitStripeCompactionRequest(sfs,
351 si.getStartRow(bestStart), si.getEndRow(endIndex), 1, Long.MAX_VALUE);
352 if (canDropDeletesNoL0) {
353 result.setMajorRangeFull();
354 }
355 return result;
356 }
357
358 private static long getTotalKvCount(final Collection<StoreFile> candidates) {
359 long totalSize = 0;
360 for (StoreFile storeFile : candidates) {
361 totalSize += storeFile.getReader().getEntries();
362 }
363 return totalSize;
364 }
365
366 public static long getTotalFileSize(final Collection<StoreFile> candidates) {
367 long totalSize = 0;
368 for (StoreFile storeFile : candidates) {
369 totalSize += storeFile.getReader().length();
370 }
371 return totalSize;
372 }
373
374 private Pair<Long, Integer> estimateTargetKvs(Collection<StoreFile> files, double splitCount) {
375
376
377
378
379 long totalSize = getTotalFileSize(files);
380 long targetPartSize = config.getSplitPartSize();
381 assert targetPartSize > 0 && splitCount > 0;
382 double ratio = totalSize / (splitCount * targetPartSize);
383 while (ratio > 1.0) {
384
385 double newRatio = totalSize / ((splitCount + 1.0) * targetPartSize);
386 if ((1.0 / newRatio) >= ratio) break;
387 ratio = newRatio;
388 splitCount += 1.0;
389 }
390 long kvCount = (long)(getTotalKvCount(files) / splitCount);
391 return new Pair<Long, Integer>(kvCount, (int)Math.ceil(splitCount));
392 }
393
394
395 public abstract static class StripeCompactionRequest {
396 protected CompactionRequest request;
397 protected byte[] majorRangeFromRow = null, majorRangeToRow = null;
398
399 public List<Path> execute(StripeCompactor compactor,
400 ThroughputController throughputController) throws IOException {
401 return execute(compactor, throughputController, null);
402 }
403
404
405
406
407
408
409 public abstract List<Path> execute(StripeCompactor compactor,
410 ThroughputController throughputController, User user) throws IOException;
411
412 public StripeCompactionRequest(CompactionRequest request) {
413 this.request = request;
414 }
415
416
417
418
419
420
421
422 public void setMajorRange(byte[] startRow, byte[] endRow) {
423 this.majorRangeFromRow = startRow;
424 this.majorRangeToRow = endRow;
425 }
426
427 public CompactionRequest getRequest() {
428 return this.request;
429 }
430
431 public void setRequest(CompactionRequest request) {
432 assert request != null;
433 this.request = request;
434 this.majorRangeFromRow = this.majorRangeToRow = null;
435 }
436 }
437
438
439
440
441
442 private static class BoundaryStripeCompactionRequest extends StripeCompactionRequest {
443 private final List<byte[]> targetBoundaries;
444
445
446
447
448
449 public BoundaryStripeCompactionRequest(CompactionRequest request,
450 List<byte[]> targetBoundaries) {
451 super(request);
452 this.targetBoundaries = targetBoundaries;
453 }
454
455 public BoundaryStripeCompactionRequest(Collection<StoreFile> files,
456 List<byte[]> targetBoundaries) {
457 this(new CompactionRequest(files), targetBoundaries);
458 }
459
460 @Override
461 public List<Path> execute(StripeCompactor compactor,
462 ThroughputController throughputController, User user) throws IOException {
463 return compactor.compact(this.request, this.targetBoundaries, this.majorRangeFromRow,
464 this.majorRangeToRow, throughputController, user);
465 }
466 }
467
468
469
470
471
472
473
474 private static class SplitStripeCompactionRequest extends StripeCompactionRequest {
475 private final byte[] startRow, endRow;
476 private final int targetCount;
477 private final long targetKvs;
478
479
480
481
482
483
484
485
486
487 public SplitStripeCompactionRequest(CompactionRequest request,
488 byte[] startRow, byte[] endRow, int targetCount, long targetKvs) {
489 super(request);
490 this.startRow = startRow;
491 this.endRow = endRow;
492 this.targetCount = targetCount;
493 this.targetKvs = targetKvs;
494 }
495
496 public SplitStripeCompactionRequest(
497 CompactionRequest request, byte[] startRow, byte[] endRow, long targetKvs) {
498 this(request, startRow, endRow, Integer.MAX_VALUE, targetKvs);
499 }
500
501 public SplitStripeCompactionRequest(
502 Collection<StoreFile> files, byte[] startRow, byte[] endRow, long targetKvs) {
503 this(files, startRow, endRow, Integer.MAX_VALUE, targetKvs);
504 }
505
506 public SplitStripeCompactionRequest(Collection<StoreFile> files,
507 byte[] startRow, byte[] endRow, int targetCount, long targetKvs) {
508 this(new CompactionRequest(files), startRow, endRow, targetCount, targetKvs);
509 }
510
511 @Override
512 public List<Path> execute(StripeCompactor compactor,
513 ThroughputController throughputController, User user) throws IOException {
514 return compactor.compact(this.request, this.targetCount, this.targetKvs, this.startRow,
515 this.endRow, this.majorRangeFromRow, this.majorRangeToRow, throughputController, user);
516 }
517
518
519
520 public void setMajorRangeFull() {
521 setMajorRange(this.startRow, this.endRow);
522 }
523 }
524
525
526 public static interface StripeInformationProvider {
527 public Collection<StoreFile> getStorefiles();
528
529
530
531
532
533
534 public byte[] getStartRow(int stripeIndex);
535
536
537
538
539
540
541 public byte[] getEndRow(int stripeIndex);
542
543
544
545
546 public List<StoreFile> getLevel0Files();
547
548
549
550
551 public List<byte[]> getStripeBoundaries();
552
553
554
555
556 public ArrayList<ImmutableList<StoreFile>> getStripes();
557
558
559
560
561 public int getStripeCount();
562 }
563 }