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;
20
21 import java.io.FileNotFoundException;
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.Date;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.NavigableSet;
28 import java.util.UUID;
29 import java.util.concurrent.ConcurrentHashMap;
30
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.hadoop.classification.InterfaceAudience;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.fs.FileSystem;
36 import org.apache.hadoop.fs.Path;
37 import org.apache.hadoop.hbase.Cell;
38 import org.apache.hadoop.hbase.HColumnDescriptor;
39 import org.apache.hadoop.hbase.HConstants;
40 import org.apache.hadoop.hbase.KeyValue;
41 import org.apache.hadoop.hbase.KeyValue.KVComparator;
42 import org.apache.hadoop.hbase.KeyValue.Type;
43 import org.apache.hadoop.hbase.TableName;
44 import org.apache.hadoop.hbase.Tag;
45 import org.apache.hadoop.hbase.client.Scan;
46 import org.apache.hadoop.hbase.filter.Filter;
47 import org.apache.hadoop.hbase.filter.FilterList;
48 import org.apache.hadoop.hbase.io.compress.Compression;
49 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
50 import org.apache.hadoop.hbase.io.hfile.HFile;
51 import org.apache.hadoop.hbase.io.hfile.CorruptHFileException;
52 import org.apache.hadoop.hbase.io.hfile.HFileContext;
53 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
54 import org.apache.hadoop.hbase.master.TableLockManager;
55 import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
56 import org.apache.hadoop.hbase.mob.MobCacheConfig;
57 import org.apache.hadoop.hbase.mob.MobConstants;
58 import org.apache.hadoop.hbase.mob.MobFile;
59 import org.apache.hadoop.hbase.mob.MobFileName;
60 import org.apache.hadoop.hbase.mob.MobStoreEngine;
61 import org.apache.hadoop.hbase.mob.MobUtils;
62 import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
63 import org.apache.hadoop.hbase.regionserver.compactions.CompactionThroughputController;
64 import org.apache.hadoop.hbase.util.Bytes;
65 import org.apache.hadoop.hbase.util.ChecksumType;
66 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
67 import org.apache.hadoop.hbase.util.IdLock;
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85 @InterfaceAudience.Private
86 public class HMobStore extends HStore {
87
88 private static final Log LOG = LogFactory.getLog(HMobStore.class);
89 private MobCacheConfig mobCacheConfig;
90 private Path homePath;
91 private Path mobFamilyPath;
92 private volatile long mobCompactedIntoMobCellsCount = 0;
93 private volatile long mobCompactedFromMobCellsCount = 0;
94 private volatile long mobCompactedIntoMobCellsSize = 0;
95 private volatile long mobCompactedFromMobCellsSize = 0;
96 private volatile long mobFlushCount = 0;
97 private volatile long mobFlushedCellsCount = 0;
98 private volatile long mobFlushedCellsSize = 0;
99 private volatile long mobScanCellsCount = 0;
100 private volatile long mobScanCellsSize = 0;
101 private HColumnDescriptor family;
102 private TableLockManager tableLockManager;
103 private TableName tableLockName;
104 private Map<String, List<Path>> map = new ConcurrentHashMap<String, List<Path>>();
105 private final IdLock keyLock = new IdLock();
106
107 public HMobStore(final HRegion region, final HColumnDescriptor family,
108 final Configuration confParam) throws IOException {
109 super(region, family, confParam);
110 this.family = family;
111 this.mobCacheConfig = (MobCacheConfig) cacheConf;
112 this.homePath = MobUtils.getMobHome(conf);
113 this.mobFamilyPath = MobUtils.getMobFamilyPath(conf, this.getTableName(),
114 family.getNameAsString());
115 List<Path> locations = new ArrayList<Path>(2);
116 locations.add(mobFamilyPath);
117 TableName tn = region.getTableDesc().getTableName();
118 locations.add(HFileArchiveUtil.getStoreArchivePath(conf, tn, MobUtils.getMobRegionInfo(tn)
119 .getEncodedName(), family.getNameAsString()));
120 map.put(Bytes.toString(tn.getName()), locations);
121 if (region.getRegionServerServices() != null) {
122 tableLockManager = region.getRegionServerServices().getTableLockManager();
123 tableLockName = MobUtils.getTableLockName(getTableName());
124 }
125 }
126
127
128
129
130 @Override
131 protected void createCacheConf(HColumnDescriptor family) {
132 cacheConf = new MobCacheConfig(conf, family);
133 }
134
135
136
137
138 public Configuration getConfiguration() {
139 return this.conf;
140 }
141
142
143
144
145
146 @Override
147 protected KeyValueScanner createScanner(Scan scan, final NavigableSet<byte[]> targetCols,
148 long readPt, KeyValueScanner scanner) throws IOException {
149 if (scanner == null) {
150 if (MobUtils.isRefOnlyScan(scan)) {
151 Filter refOnlyFilter = new MobReferenceOnlyFilter();
152 Filter filter = scan.getFilter();
153 if (filter != null) {
154 scan.setFilter(new FilterList(filter, refOnlyFilter));
155 } else {
156 scan.setFilter(refOnlyFilter);
157 }
158 }
159 scanner = scan.isReversed() ? new ReversedMobStoreScanner(this, getScanInfo(), scan,
160 targetCols, readPt) : new MobStoreScanner(this, getScanInfo(), scan, targetCols, readPt);
161 }
162 return scanner;
163 }
164
165
166
167
168 @Override
169 protected StoreEngine<?, ?, ?, ?> createStoreEngine(Store store, Configuration conf,
170 KVComparator kvComparator) throws IOException {
171 MobStoreEngine engine = new MobStoreEngine();
172 engine.createComponents(conf, store, kvComparator);
173 return engine;
174 }
175
176
177
178
179
180 private Path getTempDir() {
181 return new Path(homePath, MobConstants.TEMP_DIR_NAME);
182 }
183
184
185
186
187
188
189
190
191
192
193 public StoreFile.Writer createWriterInTmp(Date date, long maxKeyCount,
194 Compression.Algorithm compression, byte[] startKey) throws IOException {
195 if (startKey == null) {
196 startKey = HConstants.EMPTY_START_ROW;
197 }
198 Path path = getTempDir();
199 return createWriterInTmp(MobUtils.formatDate(date), path, maxKeyCount, compression, startKey);
200 }
201
202
203
204
205
206
207
208
209
210
211
212
213 public StoreFile.Writer createDelFileWriterInTmp(Date date, long maxKeyCount,
214 Compression.Algorithm compression, byte[] startKey) throws IOException {
215 if (startKey == null) {
216 startKey = HConstants.EMPTY_START_ROW;
217 }
218 Path path = getTempDir();
219 String suffix = UUID
220 .randomUUID().toString().replaceAll("-", "") + "_del";
221 MobFileName mobFileName = MobFileName.create(startKey, MobUtils.formatDate(date), suffix);
222 return createWriterInTmp(mobFileName, path, maxKeyCount, compression);
223 }
224
225
226
227
228
229
230
231
232
233
234
235 public StoreFile.Writer createWriterInTmp(String date, Path basePath, long maxKeyCount,
236 Compression.Algorithm compression, byte[] startKey) throws IOException {
237 MobFileName mobFileName = MobFileName.create(startKey, date, UUID.randomUUID()
238 .toString().replaceAll("-", ""));
239 return createWriterInTmp(mobFileName, basePath, maxKeyCount, compression);
240 }
241
242
243
244
245
246
247
248
249
250
251 public StoreFile.Writer createWriterInTmp(MobFileName mobFileName, Path basePath, long maxKeyCount,
252 Compression.Algorithm compression) throws IOException {
253 final CacheConfig writerCacheConf = mobCacheConfig;
254 HFileContext hFileContext = new HFileContextBuilder().withCompression(compression)
255 .withIncludesMvcc(true).withIncludesTags(true)
256 .withChecksumType(ChecksumType.getDefaultChecksumType())
257 .withBytesPerCheckSum(HFile.DEFAULT_BYTES_PER_CHECKSUM)
258 .withBlockSize(getFamily().getBlocksize())
259 .withHBaseCheckSum(true).withDataBlockEncoding(getFamily().getDataBlockEncoding()).build();
260
261 StoreFile.Writer w = new StoreFile.WriterBuilder(conf, writerCacheConf, region.getFilesystem())
262 .withFilePath(new Path(basePath, mobFileName.getFileName()))
263 .withComparator(KeyValue.COMPARATOR).withBloomType(BloomType.NONE)
264 .withMaxKeyCount(maxKeyCount).withFileContext(hFileContext).build();
265 return w;
266 }
267
268
269
270
271
272
273
274 public void commitFile(final Path sourceFile, Path targetPath) throws IOException {
275 if (sourceFile == null) {
276 return;
277 }
278 Path dstPath = new Path(targetPath, sourceFile.getName());
279 validateMobFile(sourceFile);
280 String msg = "Renaming flushed file from " + sourceFile + " to " + dstPath;
281 LOG.info(msg);
282 Path parent = dstPath.getParent();
283 if (!region.getFilesystem().exists(parent)) {
284 region.getFilesystem().mkdirs(parent);
285 }
286 if (!region.getFilesystem().rename(sourceFile, dstPath)) {
287 throw new IOException("Failed rename of " + sourceFile + " to " + dstPath);
288 }
289 }
290
291
292
293
294
295
296 private void validateMobFile(Path path) throws IOException {
297 StoreFile storeFile = null;
298 try {
299 storeFile =
300 new StoreFile(region.getFilesystem(), path, conf, this.mobCacheConfig, BloomType.NONE);
301 storeFile.createReader();
302 } catch (IOException e) {
303 LOG.error("Fail to open mob file[" + path + "], keep it in temp directory.", e);
304 throw e;
305 } finally {
306 if (storeFile != null) {
307 storeFile.closeReader(false);
308 }
309 }
310 }
311
312
313
314
315
316
317
318
319
320 public Cell resolve(Cell reference, boolean cacheBlocks) throws IOException {
321 return resolve(reference, cacheBlocks, -1, true);
322 }
323
324
325
326
327
328
329
330
331
332
333
334 public Cell resolve(Cell reference, boolean cacheBlocks, long readPt,
335 boolean readEmptyValueOnMobCellMiss) throws IOException {
336 Cell result = null;
337 if (MobUtils.hasValidMobRefCellValue(reference)) {
338 String fileName = MobUtils.getMobFileName(reference);
339 Tag tableNameTag = MobUtils.getTableNameTag(reference);
340 if (tableNameTag != null) {
341 byte[] tableName = tableNameTag.getValue();
342 String tableNameString = Bytes.toString(tableName);
343 List<Path> locations = map.get(tableNameString);
344 if (locations == null) {
345 IdLock.Entry lockEntry = keyLock.getLockEntry(tableNameString.hashCode());
346 try {
347 locations = map.get(tableNameString);
348 if (locations == null) {
349 locations = new ArrayList<Path>(2);
350 TableName tn = TableName.valueOf(tableName);
351 locations.add(MobUtils.getMobFamilyPath(conf, tn, family.getNameAsString()));
352 locations.add(HFileArchiveUtil.getStoreArchivePath(conf, tn, MobUtils
353 .getMobRegionInfo(tn).getEncodedName(), family.getNameAsString()));
354 map.put(tableNameString, locations);
355 }
356 } finally {
357 keyLock.releaseLockEntry(lockEntry);
358 }
359 }
360 result = readCell(locations, fileName, reference, cacheBlocks, readPt,
361 readEmptyValueOnMobCellMiss);
362 }
363 }
364 if (result == null) {
365 LOG.warn("The KeyValue result is null, assemble a new KeyValue with the same row,family,"
366 + "qualifier,timestamp,type and tags but with an empty value to return.");
367 result = new KeyValue(reference.getRowArray(), reference.getRowOffset(),
368 reference.getRowLength(), reference.getFamilyArray(), reference.getFamilyOffset(),
369 reference.getFamilyLength(), reference.getQualifierArray(),
370 reference.getQualifierOffset(), reference.getQualifierLength(), reference.getTimestamp(),
371 Type.codeToType(reference.getTypeByte()), HConstants.EMPTY_BYTE_ARRAY,
372 0, 0, reference.getTagsArray(), reference.getTagsOffset(),
373 reference.getTagsLength());
374 }
375 return result;
376 }
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394 private Cell readCell(List<Path> locations, String fileName, Cell search, boolean cacheMobBlocks,
395 long readPt, boolean readEmptyValueOnMobCellMiss) throws IOException {
396 FileSystem fs = getFileSystem();
397 Throwable throwable = null;
398 for (Path location : locations) {
399 MobFile file = null;
400 Path path = new Path(location, fileName);
401 try {
402 file = mobCacheConfig.getMobFileCache().openFile(fs, path, mobCacheConfig);
403 return readPt != -1 ? file.readCell(search, cacheMobBlocks, readPt) : file.readCell(search,
404 cacheMobBlocks);
405 } catch (IOException e) {
406 mobCacheConfig.getMobFileCache().evictFile(fileName);
407 throwable = e;
408 if ((e instanceof FileNotFoundException) ||
409 (e.getCause() instanceof FileNotFoundException)) {
410 LOG.warn("Fail to read the cell, the mob file " + path + " doesn't exist", e);
411 } else if (e instanceof CorruptHFileException) {
412 LOG.error("The mob file " + path + " is corrupt", e);
413 break;
414 } else {
415 throw e;
416 }
417 } catch (NullPointerException e) {
418 mobCacheConfig.getMobFileCache().evictFile(fileName);
419 LOG.warn("Fail to read the cell", e);
420 throwable = e;
421 } catch (AssertionError e) {
422 mobCacheConfig.getMobFileCache().evictFile(fileName);
423 LOG.warn("Fail to read the cell", e);
424 throwable = e;
425 } finally {
426 if (file != null) {
427 mobCacheConfig.getMobFileCache().closeFile(file);
428 }
429 }
430 }
431 LOG.error("The mob file " + fileName + " could not be found in the locations " + locations
432 + " or it is corrupt");
433 if (readEmptyValueOnMobCellMiss) {
434 return null;
435 } else if (throwable instanceof IOException) {
436 throw (IOException) throwable;
437 } else {
438 throw new IOException(throwable);
439 }
440 }
441
442
443
444
445
446 public Path getPath() {
447 return mobFamilyPath;
448 }
449
450
451
452
453
454
455
456
457
458
459 @Override
460 public List<StoreFile> compact(CompactionContext compaction,
461 CompactionThroughputController throughputController) throws IOException {
462
463
464 if (compaction.getRequest().isAllFiles()) {
465
466
467
468
469
470
471
472
473 TableLock lock = null;
474 if (tableLockManager != null) {
475 lock = tableLockManager.readLock(tableLockName, "Major compaction in HMobStore");
476 }
477 boolean tableLocked = false;
478 String tableName = getTableName().getNameAsString();
479 if (lock != null) {
480 try {
481 LOG.info("Start to acquire a read lock for the table[" + tableName
482 + "], ready to perform the major compaction");
483 lock.acquire();
484 tableLocked = true;
485 } catch (Exception e) {
486 LOG.error("Fail to lock the table " + tableName, e);
487 }
488 } else {
489
490 tableLocked = true;
491 }
492 try {
493 if (!tableLocked) {
494 LOG.warn("Cannot obtain the table lock, maybe a sweep tool is running on this table["
495 + tableName + "], forcing the delete markers to be retained");
496 compaction.getRequest().forceRetainDeleteMarkers();
497 }
498 return super.compact(compaction, throughputController);
499 } finally {
500 if (tableLocked && lock != null) {
501 try {
502 lock.release();
503 } catch (IOException e) {
504 LOG.error("Fail to release the table lock " + tableName, e);
505 }
506 }
507 }
508 } else {
509
510 return super.compact(compaction, throughputController);
511 }
512 }
513
514 public void updateMobCompactedIntoMobCellsCount(long count) {
515 mobCompactedIntoMobCellsCount += count;
516 }
517
518 public long getMobCompactedIntoMobCellsCount() {
519 return mobCompactedIntoMobCellsCount;
520 }
521
522 public void updateMobCompactedFromMobCellsCount(long count) {
523 mobCompactedFromMobCellsCount += count;
524 }
525
526 public long getMobCompactedFromMobCellsCount() {
527 return mobCompactedFromMobCellsCount;
528 }
529
530 public void updateMobCompactedIntoMobCellsSize(long size) {
531 mobCompactedIntoMobCellsSize += size;
532 }
533
534 public long getMobCompactedIntoMobCellsSize() {
535 return mobCompactedIntoMobCellsSize;
536 }
537
538 public void updateMobCompactedFromMobCellsSize(long size) {
539 mobCompactedFromMobCellsSize += size;
540 }
541
542 public long getMobCompactedFromMobCellsSize() {
543 return mobCompactedFromMobCellsSize;
544 }
545
546 public void updateMobFlushCount() {
547 mobFlushCount++;
548 }
549
550 public long getMobFlushCount() {
551 return mobFlushCount;
552 }
553
554 public void updateMobFlushedCellsCount(long count) {
555 mobFlushedCellsCount += count;
556 }
557
558 public long getMobFlushedCellsCount() {
559 return mobFlushedCellsCount;
560 }
561
562 public void updateMobFlushedCellsSize(long size) {
563 mobFlushedCellsSize += size;
564 }
565
566 public long getMobFlushedCellsSize() {
567 return mobFlushedCellsSize;
568 }
569
570 public void updateMobScanCellsCount(long count) {
571 mobScanCellsCount += count;
572 }
573
574 public long getMobScanCellsCount() {
575 return mobScanCellsCount;
576 }
577
578 public void updateMobScanCellsSize(long size) {
579 mobScanCellsSize += size;
580 }
581
582 public long getMobScanCellsSize() {
583 return mobScanCellsSize;
584 }
585 }