1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.regionserver.compactions;
19
20 import java.io.IOException;
21 import java.io.InterruptedIOException;
22 import java.util.ArrayList;
23 import java.util.Collection;
24 import java.util.Collections;
25 import java.util.List;
26
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.hadoop.hbase.classification.InterfaceAudience;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.fs.Path;
32 import org.apache.hadoop.hbase.regionserver.InternalScanner;
33 import org.apache.hadoop.hbase.regionserver.ScanType;
34 import org.apache.hadoop.hbase.regionserver.Store;
35 import org.apache.hadoop.hbase.regionserver.StoreFile;
36 import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
37 import org.apache.hadoop.hbase.security.User;
38
39
40
41
42
43 @InterfaceAudience.Private
44 public class DefaultCompactor extends Compactor {
45 private static final Log LOG = LogFactory.getLog(DefaultCompactor.class);
46
47 public DefaultCompactor(final Configuration conf, final Store store) {
48 super(conf, store);
49 }
50
51
52
53
54 public List<Path> compact(final CompactionRequest request,
55 CompactionThroughputController throughputController, User user) throws IOException {
56 FileDetails fd = getFileDetails(request.getFiles(), request.isAllFiles());
57 this.progress = new CompactionProgress(fd.maxKeyCount);
58
59
60 long smallestReadPoint = getSmallestReadPoint();
61
62 List<StoreFileScanner> scanners;
63 Collection<StoreFile> readersToClose;
64 if (this.conf.getBoolean("hbase.regionserver.compaction.private.readers", true)) {
65
66
67 readersToClose = new ArrayList<StoreFile>(request.getFiles().size());
68 for (StoreFile f : request.getFiles()) {
69 readersToClose.add(new StoreFile(f));
70 }
71 scanners = createFileScanners(readersToClose, smallestReadPoint,
72 store.throttleCompaction(request.getSize()));
73 } else {
74 readersToClose = Collections.emptyList();
75 scanners = createFileScanners(request.getFiles(), smallestReadPoint,
76 store.throttleCompaction(request.getSize()));
77 }
78
79 StoreFile.Writer writer = null;
80 List<Path> newFiles = new ArrayList<Path>();
81 boolean cleanSeqId = false;
82 IOException e = null;
83 try {
84 InternalScanner scanner = null;
85 try {
86
87 ScanType scanType = request.isRetainDeleteMarkers() ? ScanType.COMPACT_RETAIN_DELETES
88 : ScanType.COMPACT_DROP_DELETES;
89 scanner = preCreateCoprocScanner(request, scanType, fd.earliestPutTs, scanners, user);
90 if (scanner == null) {
91 scanner = createScanner(store, scanners, scanType, smallestReadPoint, fd.earliestPutTs);
92 }
93 scanner = postCreateCoprocScanner(request, scanType, scanner, user);
94 if (scanner == null) {
95
96 return newFiles;
97 }
98
99
100 if(fd.minSeqIdToKeep > 0) {
101 smallestReadPoint = Math.min(fd.minSeqIdToKeep, smallestReadPoint);
102 cleanSeqId = true;
103 }
104
105 writer = createTmpWriter(fd, store.throttleCompaction(request.getSize()));
106 boolean finished = performCompaction(fd, scanner, writer, smallestReadPoint, cleanSeqId,
107 throughputController, request.isAllFiles());
108
109 if (!finished) {
110 writer.close();
111 store.getFileSystem().delete(writer.getPath(), false);
112 writer = null;
113 throw new InterruptedIOException("Aborting compaction of store " + store +
114 " in region " + store.getRegionInfo().getRegionNameAsString() +
115 " because it was interrupted.");
116 }
117 } finally {
118 if (scanner != null) {
119 scanner.close();
120 }
121 }
122 } catch (IOException ioe) {
123 e = ioe;
124
125 throw ioe;
126 }
127 finally {
128 try {
129 if (writer != null) {
130 if (e != null) {
131 writer.close();
132 } else {
133 writer.appendMetadata(fd.maxSeqId, request.isAllFiles());
134 writer.close();
135 newFiles.add(writer.getPath());
136 }
137 }
138 } finally {
139 for (StoreFile f : readersToClose) {
140 try {
141 f.closeReader(true);
142 } catch (IOException ioe) {
143 LOG.warn("Exception closing " + f, ioe);
144 }
145 }
146 }
147 }
148 return newFiles;
149 }
150
151
152
153
154
155
156
157 protected StoreFile.Writer createTmpWriter(FileDetails fd, boolean shouldDropBehind)
158 throws IOException {
159
160
161
162
163 return store.createWriterInTmp(fd.maxKeyCount, this.compactionCompression,
164
165
166
167
168 }
169
170
171
172
173
174
175
176
177
178
179
180 public List<Path> compactForTesting(final Collection<StoreFile> filesToCompact, boolean isMajor)
181 throws IOException {
182 CompactionRequest cr = new CompactionRequest(filesToCompact);
183 cr.setIsMajor(isMajor, isMajor);
184 return this.compact(cr, NoLimitCompactionThroughputController.INSTANCE, null);
185 }
186 }