1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.mob;
19
20 import java.io.IOException;
21 import java.util.ArrayList;
22 import java.util.Date;
23 import java.util.List;
24
25 import org.apache.commons.logging.Log;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.hadoop.classification.InterfaceAudience;
28 import org.apache.hadoop.conf.Configuration;
29 import org.apache.hadoop.fs.Path;
30 import org.apache.hadoop.hbase.Cell;
31 import org.apache.hadoop.hbase.CellUtil;
32 import org.apache.hadoop.hbase.KeyValue;
33 import org.apache.hadoop.hbase.KeyValueUtil;
34 import org.apache.hadoop.hbase.Tag;
35 import org.apache.hadoop.hbase.TagType;
36 import org.apache.hadoop.hbase.client.Scan;
37 import org.apache.hadoop.hbase.regionserver.HMobStore;
38 import org.apache.hadoop.hbase.regionserver.HStore;
39 import org.apache.hadoop.hbase.regionserver.InternalScanner;
40 import org.apache.hadoop.hbase.regionserver.MobCompactionStoreScanner;
41 import org.apache.hadoop.hbase.regionserver.ScanType;
42 import org.apache.hadoop.hbase.regionserver.ScannerContext;
43 import org.apache.hadoop.hbase.regionserver.Store;
44 import org.apache.hadoop.hbase.regionserver.StoreFile;
45 import org.apache.hadoop.hbase.regionserver.StoreFile.Writer;
46 import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
47 import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
48 import org.apache.hadoop.hbase.regionserver.compactions.CompactionThroughputController;
49 import org.apache.hadoop.hbase.util.Bytes;
50
51
52
53
54 @InterfaceAudience.Private
55 public class DefaultMobCompactor extends DefaultCompactor {
56
57 private static final Log LOG = LogFactory.getLog(DefaultMobCompactor.class);
58 private long mobSizeThreshold;
59 private HMobStore mobStore;
60 public DefaultMobCompactor(Configuration conf, Store store) {
61 super(conf, store);
62
63
64
65
66 if (!(store instanceof HMobStore)) {
67 throw new IllegalArgumentException("The store " + store + " is not a HMobStore");
68 }
69 mobStore = (HMobStore) store;
70 mobSizeThreshold = store.getFamily().getMobThreshold();
71 }
72
73
74
75
76
77
78
79
80 @Override
81 protected Writer createTmpWriter(FileDetails fd, boolean shouldDropBehind) throws IOException {
82
83 StoreFile.Writer writer = store.createWriterInTmp(fd.maxKeyCount, this.compactionCompression,
84 true, true, true, shouldDropBehind);
85 return writer;
86 }
87
88 @Override
89 protected InternalScanner createScanner(Store store, List<StoreFileScanner> scanners,
90 ScanType scanType, long smallestReadPoint, long earliestPutTs) throws IOException {
91 Scan scan = new Scan();
92 scan.setMaxVersions(store.getFamily().getMaxVersions());
93 if (scanType == ScanType.COMPACT_DROP_DELETES) {
94 scanType = ScanType.COMPACT_RETAIN_DELETES;
95 return new MobCompactionStoreScanner(store, store.getScanInfo(), scan, scanners,
96 scanType, smallestReadPoint, earliestPutTs, true);
97 } else {
98 return new MobCompactionStoreScanner(store, store.getScanInfo(), scan, scanners,
99 scanType, smallestReadPoint, earliestPutTs, false);
100 }
101 }
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145 @Override
146 protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
147 long smallestReadPoint, boolean cleanSeqId,
148 CompactionThroughputController throughputController, boolean major) throws IOException {
149 if (!(scanner instanceof MobCompactionStoreScanner)) {
150 throw new IllegalArgumentException(
151 "The scanner should be an instance of MobCompactionStoreScanner");
152 }
153 MobCompactionStoreScanner compactionScanner = (MobCompactionStoreScanner) scanner;
154 int bytesWritten = 0;
155
156
157 List<Cell> cells = new ArrayList<Cell>();
158
159 int closeCheckInterval = HStore.getCloseCheckInterval();
160 boolean hasMore;
161 Path path = MobUtils.getMobFamilyPath(conf, store.getTableName(), store.getColumnFamilyName());
162 byte[] fileName = null;
163 StoreFile.Writer mobFileWriter = null;
164 StoreFile.Writer delFileWriter = null;
165 long mobCells = 0;
166 long deleteMarkersCount = 0;
167 Tag tableNameTag = new Tag(TagType.MOB_TABLE_NAME_TAG_TYPE, store.getTableName()
168 .getName());
169 long mobCompactedIntoMobCellsCount = 0;
170 long mobCompactedFromMobCellsCount = 0;
171 long mobCompactedIntoMobCellsSize = 0;
172 long mobCompactedFromMobCellsSize = 0;
173 try {
174 try {
175
176 mobFileWriter = mobStore.createWriterInTmp(new Date(fd.latestPutTs), fd.maxKeyCount,
177 store.getFamily().getCompression(), store.getRegionInfo().getStartKey());
178 fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
179 } catch (IOException e) {
180 LOG.error(
181 "Fail to create mob writer, "
182 + "we will continue the compaction by writing MOB cells directly in store files",
183 e);
184 }
185 delFileWriter = mobStore.createDelFileWriterInTmp(new Date(fd.latestPutTs), fd.maxKeyCount,
186 store.getFamily().getCompression(), store.getRegionInfo().getStartKey());
187 ScannerContext scannerContext =
188 ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
189 do {
190 hasMore = compactionScanner.next(cells, scannerContext);
191
192 for (Cell c : cells) {
193
194 KeyValue kv = KeyValueUtil.ensureKeyValue(c);
195 resetSeqId(smallestReadPoint, cleanSeqId, kv);
196 if (compactionScanner.isOutputDeleteMarkers() && CellUtil.isDelete(c)) {
197 delFileWriter.append(kv);
198 deleteMarkersCount++;
199 } else if (mobFileWriter == null || kv.getTypeByte() != KeyValue.Type.Put.getCode()) {
200
201
202 writer.append(kv);
203 } else if (MobUtils.isMobReferenceCell(kv)) {
204 if (MobUtils.hasValidMobRefCellValue(kv)) {
205 int size = MobUtils.getMobValueLength(kv);
206 if (size > mobSizeThreshold) {
207
208
209 writer.append(kv);
210 } else {
211
212
213 Cell cell = mobStore.resolve(kv, false);
214 if (cell.getValueLength() != 0) {
215
216 KeyValue mobKv = KeyValueUtil.ensureKeyValue(cell);
217 mobKv.setSequenceId(kv.getSequenceId());
218 writer.append(mobKv);
219 mobCompactedFromMobCellsCount++;
220 mobCompactedFromMobCellsSize += cell.getValueLength();
221 } else {
222
223
224
225 writer.append(kv);
226 }
227 }
228 } else {
229 LOG.warn("The value format of the KeyValue " + kv
230 + " is wrong, its length is less than " + Bytes.SIZEOF_INT);
231 writer.append(kv);
232 }
233 } else if (kv.getValueLength() <= mobSizeThreshold) {
234
235
236 writer.append(kv);
237 } else {
238
239
240 mobCells++;
241
242 mobFileWriter.append(kv);
243 KeyValue reference = MobUtils.createMobRefKeyValue(kv, fileName, tableNameTag);
244
245 writer.append(reference);
246 mobCompactedIntoMobCellsCount++;
247 mobCompactedIntoMobCellsSize += kv.getValueLength();
248 }
249 ++progress.currentCompactedKVs;
250
251
252 if (closeCheckInterval > 0) {
253 bytesWritten += kv.getLength();
254 if (bytesWritten > closeCheckInterval) {
255 bytesWritten = 0;
256 if (!store.areWritesEnabled()) {
257 progress.cancel();
258 return false;
259 }
260 }
261 }
262 }
263 cells.clear();
264 } while (hasMore);
265 } finally {
266 if (mobFileWriter != null) {
267 mobFileWriter.appendMetadata(fd.maxSeqId, major, mobCells);
268 mobFileWriter.close();
269 }
270 if (delFileWriter != null) {
271 delFileWriter.appendMetadata(fd.maxSeqId, major, deleteMarkersCount);
272 delFileWriter.close();
273 }
274 }
275 if (mobFileWriter != null) {
276 if (mobCells > 0) {
277
278 mobStore.commitFile(mobFileWriter.getPath(), path);
279 } else {
280 try {
281
282 store.getFileSystem().delete(mobFileWriter.getPath(), true);
283 } catch (IOException e) {
284 LOG.error("Fail to delete the temp mob file", e);
285 }
286 }
287 }
288 if (delFileWriter != null) {
289 if (deleteMarkersCount > 0) {
290
291
292 mobStore.commitFile(delFileWriter.getPath(), path);
293 } else {
294 try {
295
296 store.getFileSystem().delete(delFileWriter.getPath(), true);
297 } catch (IOException e) {
298 LOG.error("Fail to delete the temp del file", e);
299 }
300 }
301 }
302 mobStore.updateMobCompactedFromMobCellsCount(mobCompactedFromMobCellsCount);
303 mobStore.updateMobCompactedIntoMobCellsCount(mobCompactedIntoMobCellsCount);
304 mobStore.updateMobCompactedFromMobCellsSize(mobCompactedFromMobCellsSize);
305 mobStore.updateMobCompactedIntoMobCellsSize(mobCompactedIntoMobCellsSize);
306 progress.complete();
307 return true;
308 }
309 }