1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mob;
20
21 import java.io.FileNotFoundException;
22 import java.io.IOException;
23 import java.text.ParseException;
24 import java.text.SimpleDateFormat;
25 import java.util.ArrayList;
26 import java.util.Collection;
27 import java.util.Date;
28 import java.util.List;
29 import java.util.UUID;
30 import java.util.concurrent.ExecutorService;
31 import java.util.concurrent.RejectedExecutionException;
32 import java.util.concurrent.RejectedExecutionHandler;
33 import java.util.concurrent.SynchronousQueue;
34 import java.util.concurrent.ThreadPoolExecutor;
35 import java.util.concurrent.TimeUnit;
36
37 import org.apache.commons.logging.Log;
38 import org.apache.commons.logging.LogFactory;
39 import org.apache.hadoop.classification.InterfaceAudience;
40 import org.apache.hadoop.conf.Configuration;
41 import org.apache.hadoop.fs.FileStatus;
42 import org.apache.hadoop.fs.FileSystem;
43 import org.apache.hadoop.fs.Path;
44 import org.apache.hadoop.hbase.Cell;
45 import org.apache.hadoop.hbase.HBaseConfiguration;
46 import org.apache.hadoop.hbase.HColumnDescriptor;
47 import org.apache.hadoop.hbase.HConstants;
48 import org.apache.hadoop.hbase.HRegionInfo;
49 import org.apache.hadoop.hbase.KeyValue;
50 import org.apache.hadoop.hbase.TableName;
51 import org.apache.hadoop.hbase.Tag;
52 import org.apache.hadoop.hbase.TagType;
53 import org.apache.hadoop.hbase.backup.HFileArchiver;
54 import org.apache.hadoop.hbase.client.Scan;
55 import org.apache.hadoop.hbase.io.HFileLink;
56 import org.apache.hadoop.hbase.io.compress.Compression;
57 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
58 import org.apache.hadoop.hbase.io.hfile.HFile;
59 import org.apache.hadoop.hbase.io.hfile.HFileContext;
60 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
61 import org.apache.hadoop.hbase.master.TableLockManager;
62 import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
63 import org.apache.hadoop.hbase.mob.filecompactions.MobFileCompactor;
64 import org.apache.hadoop.hbase.mob.filecompactions.PartitionedMobFileCompactor;
65 import org.apache.hadoop.hbase.regionserver.BloomType;
66 import org.apache.hadoop.hbase.regionserver.HStore;
67 import org.apache.hadoop.hbase.regionserver.StoreFile;
68 import org.apache.hadoop.hbase.util.*;
69
70
71
72
73 @InterfaceAudience.Private
74 public class MobUtils {
75
76 private static final Log LOG = LogFactory.getLog(MobUtils.class);
77
78 private static final ThreadLocal<SimpleDateFormat> LOCAL_FORMAT =
79 new ThreadLocal<SimpleDateFormat>() {
80 @Override
81 protected SimpleDateFormat initialValue() {
82 return new SimpleDateFormat("yyyyMMdd");
83 }
84 };
85
86
87
88
89
90
91 public static String formatDate(Date date) {
92 return LOCAL_FORMAT.get().format(date);
93 }
94
95
96
97
98
99
100
101 public static Date parseDate(String dateString) throws ParseException {
102 return LOCAL_FORMAT.get().parse(dateString);
103 }
104
105
106
107
108
109
110 public static boolean isMobReferenceCell(Cell cell) {
111 if (cell.getTagsLength() > 0) {
112 Tag tag = Tag.getTag(cell.getTagsArray(), cell.getTagsOffset(), cell.getTagsLength(),
113 TagType.MOB_REFERENCE_TAG_TYPE);
114 return tag != null;
115 }
116 return false;
117 }
118
119
120
121
122
123
124 public static Tag getTableNameTag(Cell cell) {
125 if (cell.getTagsLength() > 0) {
126 Tag tag = Tag.getTag(cell.getTagsArray(), cell.getTagsOffset(), cell.getTagsLength(),
127 TagType.MOB_TABLE_NAME_TAG_TYPE);
128 return tag;
129 }
130 return null;
131 }
132
133
134
135
136
137
138 public static boolean hasMobReferenceTag(List<Tag> tags) {
139 if (!tags.isEmpty()) {
140 for (Tag tag : tags) {
141 if (tag.getType() == TagType.MOB_REFERENCE_TAG_TYPE) {
142 return true;
143 }
144 }
145 }
146 return false;
147 }
148
149
150
151
152
153
154
155
156
157
158 public static boolean isRawMobScan(Scan scan) {
159 byte[] raw = scan.getAttribute(MobConstants.MOB_SCAN_RAW);
160 try {
161 return raw != null && Bytes.toBoolean(raw);
162 } catch (IllegalArgumentException e) {
163 return false;
164 }
165 }
166
167
168
169
170
171
172
173
174 public static boolean isRefOnlyScan(Scan scan) {
175 byte[] refOnly = scan.getAttribute(MobConstants.MOB_SCAN_REF_ONLY);
176 try {
177 return refOnly != null && Bytes.toBoolean(refOnly);
178 } catch (IllegalArgumentException e) {
179 return false;
180 }
181 }
182
183
184
185
186
187
188
189 public static boolean isCacheMobBlocks(Scan scan) {
190 byte[] cache = scan.getAttribute(MobConstants.MOB_CACHE_BLOCKS);
191 try {
192 return cache != null && Bytes.toBoolean(cache);
193 } catch (IllegalArgumentException e) {
194 return false;
195 }
196 }
197
198
199
200
201
202
203
204
205
206
207
208 public static void setCacheMobBlocks(Scan scan, boolean cacheBlocks) {
209 scan.setAttribute(MobConstants.MOB_CACHE_BLOCKS, Bytes.toBytes(cacheBlocks));
210 }
211
212
213
214
215
216
217
218
219
220
221
222
223
224 public static void cleanExpiredMobFiles(FileSystem fs, Configuration conf, TableName tableName,
225 HColumnDescriptor columnDescriptor, CacheConfig cacheConfig, long current)
226 throws IOException {
227 long timeToLive = columnDescriptor.getTimeToLive();
228 if (Integer.MAX_VALUE == timeToLive) {
229
230 return;
231 }
232
233 Date expireDate = new Date(current - timeToLive * 1000);
234 expireDate = new Date(expireDate.getYear(), expireDate.getMonth(), expireDate.getDate());
235 LOG.info("MOB HFiles older than " + expireDate.toGMTString() + " will be deleted!");
236
237 FileStatus[] stats = null;
238 Path mobTableDir = FSUtils.getTableDir(getMobHome(conf), tableName);
239 Path path = getMobFamilyPath(conf, tableName, columnDescriptor.getNameAsString());
240 try {
241 stats = fs.listStatus(path);
242 } catch (FileNotFoundException e) {
243 LOG.warn("Fail to find the mob file " + path, e);
244 }
245 if (null == stats) {
246
247 return;
248 }
249 List<StoreFile> filesToClean = new ArrayList<StoreFile>();
250 int deletedFileCount = 0;
251 for (FileStatus file : stats) {
252 String fileName = file.getPath().getName();
253 try {
254 MobFileName mobFileName = null;
255 if (!HFileLink.isHFileLink(file.getPath())) {
256 mobFileName = MobFileName.create(fileName);
257 } else {
258 HFileLink hfileLink = HFileLink.buildFromHFileLinkPattern(conf, file.getPath());
259 mobFileName = MobFileName.create(hfileLink.getOriginPath().getName());
260 }
261 Date fileDate = parseDate(mobFileName.getDate());
262 if (LOG.isDebugEnabled()) {
263 LOG.debug("Checking file " + fileName);
264 }
265 if (fileDate.getTime() < expireDate.getTime()) {
266 if (LOG.isDebugEnabled()) {
267 LOG.debug(fileName + " is an expired file");
268 }
269 filesToClean.add(new StoreFile(fs, file.getPath(), conf, cacheConfig, BloomType.NONE));
270 }
271 } catch (Exception e) {
272 LOG.error("Cannot parse the fileName " + fileName, e);
273 }
274 }
275 if (!filesToClean.isEmpty()) {
276 try {
277 removeMobFiles(conf, fs, tableName, mobTableDir, columnDescriptor.getName(),
278 filesToClean);
279 deletedFileCount = filesToClean.size();
280 } catch (IOException e) {
281 LOG.error("Fail to delete the mob files " + filesToClean, e);
282 }
283 }
284 LOG.info(deletedFileCount + " expired mob files are deleted");
285 }
286
287
288
289
290
291
292
293 public static Path getMobHome(Configuration conf) {
294 Path hbaseDir = new Path(conf.get(HConstants.HBASE_DIR));
295 return new Path(hbaseDir, MobConstants.MOB_DIR_NAME);
296 }
297
298
299
300
301
302
303
304 public static Path getQualifiedMobRootDir(Configuration conf) throws IOException {
305 Path hbaseDir = new Path(conf.get(HConstants.HBASE_DIR));
306 Path mobRootDir = new Path(hbaseDir, MobConstants.MOB_DIR_NAME);
307 FileSystem fs = mobRootDir.getFileSystem(conf);
308 return mobRootDir.makeQualified(fs);
309 }
310
311
312
313
314
315
316
317
318 public static Path getMobRegionPath(Configuration conf, TableName tableName) {
319 Path tablePath = FSUtils.getTableDir(getMobHome(conf), tableName);
320 HRegionInfo regionInfo = getMobRegionInfo(tableName);
321 return new Path(tablePath, regionInfo.getEncodedName());
322 }
323
324
325
326
327
328
329
330
331
332 public static Path getMobFamilyPath(Configuration conf, TableName tableName, String familyName) {
333 return new Path(getMobRegionPath(conf, tableName), familyName);
334 }
335
336
337
338
339
340
341
342
343 public static Path getMobFamilyPath(Path regionPath, String familyName) {
344 return new Path(regionPath, familyName);
345 }
346
347
348
349
350
351
352
353
354 public static HRegionInfo getMobRegionInfo(TableName tableName) {
355 HRegionInfo info = new HRegionInfo(tableName, MobConstants.MOB_REGION_NAME_BYTES,
356 HConstants.EMPTY_END_ROW, false, 0);
357 return info;
358 }
359
360
361
362
363
364
365 public static boolean isMobRegionInfo(HRegionInfo regionInfo) {
366 return regionInfo == null ? false : getMobRegionInfo(regionInfo.getTable()).getEncodedName()
367 .equals(regionInfo.getEncodedName());
368 }
369
370
371
372
373
374
375
376 public static boolean isMobRegionName(TableName tableName, byte[] regionName) {
377 return Bytes.equals(regionName, getMobRegionInfo(tableName).getRegionName());
378 }
379
380
381
382
383
384
385
386 public static Path getCompactionWorkingPath(Path root, String jobName) {
387 return new Path(root, jobName);
388 }
389
390
391
392
393
394
395
396
397
398
399
400 public static void removeMobFiles(Configuration conf, FileSystem fs, TableName tableName,
401 Path tableDir, byte[] family, Collection<StoreFile> storeFiles) throws IOException {
402 HFileArchiver.archiveStoreFiles(conf, fs, getMobRegionInfo(tableName), tableDir, family,
403 storeFiles);
404 }
405
406
407
408
409
410
411
412
413
414
415 public static KeyValue createMobRefKeyValue(Cell cell, byte[] fileName, Tag tableNameTag) {
416
417
418 List<Tag> tags = new ArrayList<Tag>();
419
420 tags.add(MobConstants.MOB_REF_TAG);
421
422
423
424
425
426 tags.add(tableNameTag);
427
428 tags.addAll(Tag.asList(cell.getTagsArray(), cell.getTagsOffset(), cell.getTagsLength()));
429 int valueLength = cell.getValueLength();
430 byte[] refValue = Bytes.add(Bytes.toBytes(valueLength), fileName);
431 KeyValue reference = new KeyValue(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
432 cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
433 cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
434 cell.getTimestamp(), KeyValue.Type.Put, refValue, 0, refValue.length, tags);
435 reference.setSequenceId(cell.getSequenceId());
436 return reference;
437 }
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453 public static StoreFile.Writer createWriter(Configuration conf, FileSystem fs,
454 HColumnDescriptor family, String date, Path basePath, long maxKeyCount,
455 Compression.Algorithm compression, String startKey, CacheConfig cacheConfig)
456 throws IOException {
457 MobFileName mobFileName = MobFileName.create(startKey, date, UUID.randomUUID().toString()
458 .replaceAll("-", ""));
459 return createWriter(conf, fs, family, mobFileName, basePath, maxKeyCount, compression,
460 cacheConfig);
461 }
462
463
464
465
466
467
468
469
470
471
472
473
474 public static StoreFile.Writer createRefFileWriter(Configuration conf, FileSystem fs,
475 HColumnDescriptor family, Path basePath, long maxKeyCount, CacheConfig cacheConfig)
476 throws IOException {
477 HFileContext hFileContext = new HFileContextBuilder().withIncludesMvcc(true)
478 .withIncludesTags(true).withCompression(family.getCompactionCompression())
479 .withCompressTags(family.shouldCompressTags()).withChecksumType(HStore.getChecksumType(conf))
480 .withBytesPerCheckSum(HStore.getBytesPerChecksum(conf)).withBlockSize(family.getBlocksize())
481 .withHBaseCheckSum(true).withDataBlockEncoding(family.getDataBlockEncoding()).build();
482 Path tempPath = new Path(basePath, UUID.randomUUID().toString().replaceAll("-", ""));
483 StoreFile.Writer w = new StoreFile.WriterBuilder(conf, cacheConfig, fs).withFilePath(tempPath)
484 .withComparator(KeyValue.COMPARATOR).withBloomType(family.getBloomFilterType())
485 .withMaxKeyCount(maxKeyCount).withFileContext(hFileContext).build();
486 return w;
487 }
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503 public static StoreFile.Writer createWriter(Configuration conf, FileSystem fs,
504 HColumnDescriptor family, String date, Path basePath, long maxKeyCount,
505 Compression.Algorithm compression, byte[] startKey, CacheConfig cacheConfig)
506 throws IOException {
507 MobFileName mobFileName = MobFileName.create(startKey, date, UUID.randomUUID().toString()
508 .replaceAll("-", ""));
509 return createWriter(conf, fs, family, mobFileName, basePath, maxKeyCount, compression,
510 cacheConfig);
511 }
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527 public static StoreFile.Writer createDelFileWriter(Configuration conf, FileSystem fs,
528 HColumnDescriptor family, String date, Path basePath, long maxKeyCount,
529 Compression.Algorithm compression, byte[] startKey, CacheConfig cacheConfig)
530 throws IOException {
531 String suffix = UUID
532 .randomUUID().toString().replaceAll("-", "") + "_del";
533 MobFileName mobFileName = MobFileName.create(startKey, date, suffix);
534 return createWriter(conf, fs, family, mobFileName, basePath, maxKeyCount, compression,
535 cacheConfig);
536 }
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551 private static StoreFile.Writer createWriter(Configuration conf, FileSystem fs,
552 HColumnDescriptor family, MobFileName mobFileName, Path basePath, long maxKeyCount,
553 Compression.Algorithm compression, CacheConfig cacheConfig) throws IOException {
554 HFileContext hFileContext = new HFileContextBuilder().withCompression(compression)
555 .withIncludesMvcc(true).withIncludesTags(true).withChecksumType(
556 ChecksumType.getDefaultChecksumType())
557 .withBytesPerCheckSum(HFile.DEFAULT_BYTES_PER_CHECKSUM).withBlockSize(family.getBlocksize())
558 .withHBaseCheckSum(true).withDataBlockEncoding(family.getDataBlockEncoding()).build();
559
560 StoreFile.Writer w = new StoreFile.WriterBuilder(conf, cacheConfig, fs)
561 .withFilePath(new Path(basePath, mobFileName.getFileName()))
562 .withComparator(KeyValue.COMPARATOR).withBloomType(BloomType.NONE)
563 .withMaxKeyCount(maxKeyCount).withFileContext(hFileContext).build();
564 return w;
565 }
566
567
568
569
570
571
572
573
574
575
576
577 public static Path commitFile(Configuration conf, FileSystem fs, final Path sourceFile,
578 Path targetPath, CacheConfig cacheConfig) throws IOException {
579 if (sourceFile == null) {
580 return null;
581 }
582 Path dstPath = new Path(targetPath, sourceFile.getName());
583 validateMobFile(conf, fs, sourceFile, cacheConfig);
584 String msg = "Renaming flushed file from " + sourceFile + " to " + dstPath;
585 LOG.info(msg);
586 Path parent = dstPath.getParent();
587 if (!fs.exists(parent)) {
588 fs.mkdirs(parent);
589 }
590 if (!fs.rename(sourceFile, dstPath)) {
591 throw new IOException("Failed rename of " + sourceFile + " to " + dstPath);
592 }
593 return dstPath;
594 }
595
596
597
598
599
600
601
602
603 private static void validateMobFile(Configuration conf, FileSystem fs, Path path,
604 CacheConfig cacheConfig) throws IOException {
605 StoreFile storeFile = null;
606 try {
607 storeFile = new StoreFile(fs, path, conf, cacheConfig, BloomType.NONE);
608 storeFile.createReader();
609 } catch (IOException e) {
610 LOG.error("Fail to open mob file[" + path + "], keep it in temp directory.", e);
611 throw e;
612 } finally {
613 if (storeFile != null) {
614 storeFile.closeReader(false);
615 }
616 }
617 }
618
619
620
621
622
623
624
625
626
627
628 public static boolean hasValidMobRefCellValue(Cell cell) {
629 return cell.getValueLength() > Bytes.SIZEOF_INT;
630 }
631
632
633
634
635
636
637
638
639
640
641 public static int getMobValueLength(Cell cell) {
642 return Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), Bytes.SIZEOF_INT);
643 }
644
645
646
647
648
649
650
651
652
653
654 public static String getMobFileName(Cell cell) {
655 return Bytes.toString(cell.getValueArray(), cell.getValueOffset() + Bytes.SIZEOF_INT,
656 cell.getValueLength() - Bytes.SIZEOF_INT);
657 }
658
659
660
661
662
663
664
665 public static TableName getTableLockName(TableName tn) {
666 byte[] tableName = tn.getName();
667 return TableName.valueOf(Bytes.add(tableName, MobConstants.MOB_TABLE_LOCK_SUFFIX));
668 }
669
670
671
672
673
674
675
676
677
678
679
680 public static void doMobFileCompaction(Configuration conf, FileSystem fs, TableName tableName,
681 HColumnDescriptor hcd, ExecutorService pool, TableLockManager tableLockManager,
682 boolean isForceAllFiles) throws IOException {
683 String className = conf.get(MobConstants.MOB_FILE_COMPACTOR_CLASS_KEY,
684 PartitionedMobFileCompactor.class.getName());
685
686 MobFileCompactor compactor = null;
687 try {
688 compactor = ReflectionUtils.instantiateWithCustomCtor(className, new Class[] {
689 Configuration.class, FileSystem.class, TableName.class, HColumnDescriptor.class,
690 ExecutorService.class }, new Object[] { conf, fs, tableName, hcd, pool });
691 } catch (Exception e) {
692 throw new IOException("Unable to load configured mob file compactor '" + className + "'", e);
693 }
694
695
696
697 boolean tableLocked = false;
698 TableLock lock = null;
699 try {
700
701 if (tableLockManager != null) {
702 lock = tableLockManager.writeLock(MobUtils.getTableLockName(tableName),
703 "Run MobFileCompaction");
704 lock.acquire();
705 }
706 tableLocked = true;
707 compactor.compact(isForceAllFiles);
708 } catch (Exception e) {
709 LOG.error("Fail to compact the mob files for the column " + hcd.getNameAsString()
710 + " in the table " + tableName.getNameAsString(), e);
711 } finally {
712 if (lock != null && tableLocked) {
713 try {
714 lock.release();
715 } catch (IOException e) {
716 LOG.error("Fail to release the write lock for the table " + tableName.getNameAsString(),
717 e);
718 }
719 }
720 }
721 }
722
723
724
725
726
727
728 public static ExecutorService createMobFileCompactorThreadPool(Configuration conf) {
729 int maxThreads = conf.getInt(MobConstants.MOB_FILE_COMPACTION_THREADS_MAX,
730 MobConstants.DEFAULT_MOB_FILE_COMPACTION_THREADS_MAX);
731 if (maxThreads == 0) {
732 maxThreads = 1;
733 }
734 final SynchronousQueue<Runnable> queue = new SynchronousQueue<Runnable>();
735 ThreadPoolExecutor pool = new ThreadPoolExecutor(1, maxThreads, 60, TimeUnit.SECONDS, queue,
736 Threads.newDaemonThreadFactory("MobFileCompactor"), new RejectedExecutionHandler() {
737 @Override
738 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
739 try {
740
741 queue.put(r);
742 } catch (InterruptedException e) {
743 throw new RejectedExecutionException(e);
744 }
745 }
746 });
747 ((ThreadPoolExecutor) pool).allowCoreThreadTimeOut(true);
748 return pool;
749 }
750
751
752
753
754
755
756
757 public static boolean isReadEmptyValueOnMobCellMiss(Scan scan) {
758 byte[] readEmptyValueOnMobCellMiss = scan.getAttribute(MobConstants.EMPTY_VALUE_ON_MOBCELL_MISS);
759 try {
760 return readEmptyValueOnMobCellMiss != null && Bytes.toBoolean(readEmptyValueOnMobCellMiss);
761 } catch (IllegalArgumentException e) {
762 return false;
763 }
764 }
765
766
767
768
769
770
771
772
773
774
775 public static void archiveMobStoreFiles(Configuration conf, FileSystem fs,
776 HRegionInfo mobRegionInfo, Path mobFamilyDir, byte[] family) throws IOException {
777
778 Configuration copyOfConf = HBaseConfiguration.create(conf);
779 copyOfConf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0f);
780 CacheConfig cacheConfig = new CacheConfig(copyOfConf);
781 FileStatus[] fileStatus = FSUtils.listStatus(fs, mobFamilyDir);
782 List<StoreFile> storeFileList = new ArrayList<StoreFile>();
783 for (FileStatus file : fileStatus) {
784 storeFileList.add(new StoreFile(fs, file.getPath(), conf, cacheConfig, BloomType.NONE));
785 }
786 HFileArchiver.archiveStoreFiles(conf, fs, mobRegionInfo, mobFamilyDir, family, storeFileList);
787 }
788 }