View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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   * The mob utilities
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     * Formats a date to a string.
88     * @param date The date.
89     * @return The string format of the date, it's yyyymmdd.
90     */
91    public static String formatDate(Date date) {
92      return LOCAL_FORMAT.get().format(date);
93    }
94  
95    /**
96     * Parses the string to a date.
97     * @param dateString The string format of a date, it's yyyymmdd.
98     * @return A date.
99     * @throws ParseException
100    */
101   public static Date parseDate(String dateString) throws ParseException {
102     return LOCAL_FORMAT.get().parse(dateString);
103   }
104 
105   /**
106    * Whether the current cell is a mob reference cell.
107    * @param cell The current cell.
108    * @return True if the cell has a mob reference tag, false if it doesn't.
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    * Gets the table name tag.
121    * @param cell The current cell.
122    * @return The table name tag.
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    * Whether the tag list has a mob reference tag.
135    * @param tags The tag list.
136    * @return True if the list has a mob reference tag, false if it doesn't.
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    * Indicates whether it's a raw scan.
151    * The information is set in the attribute "hbase.mob.scan.raw" of scan.
152    * For a mob cell, in a normal scan the scanners retrieves the mob cell from the mob file.
153    * In a raw scan, the scanner directly returns cell in HBase without retrieve the one in
154    * the mob file.
155    * @param scan The current scan.
156    * @return True if it's a raw scan.
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    * Indicates whether it's a reference only scan.
169    * The information is set in the attribute "hbase.mob.scan.ref.only" of scan.
170    * If it's a ref only scan, only the cells with ref tag are returned.
171    * @param scan The current scan.
172    * @return True if it's a ref only scan.
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    * Indicates whether the scan contains the information of caching blocks.
185    * The information is set in the attribute "hbase.mob.cache.blocks" of scan.
186    * @param scan The current scan.
187    * @return True when the Scan attribute specifies to cache the MOB blocks.
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    * Sets the attribute of caching blocks in the scan.
200    *
201    * @param scan
202    *          The current scan.
203    * @param cacheBlocks
204    *          True, set the attribute of caching blocks into the scan, the scanner with this scan
205    *          caches blocks.
206    *          False, the scanner doesn't cache blocks for this scan.
207    */
208   public static void setCacheMobBlocks(Scan scan, boolean cacheBlocks) {
209     scan.setAttribute(MobConstants.MOB_CACHE_BLOCKS, Bytes.toBytes(cacheBlocks));
210   }
211 
212   /**
213    * Cleans the expired mob files.
214    * Cleans the files whose creation date is older than (current - columnFamily.ttl), and
215    * the minVersions of that column family is 0.
216    * @param fs The current file system.
217    * @param conf The current configuration.
218    * @param tableName The current table name.
219    * @param columnDescriptor The descriptor of the current column family.
220    * @param cacheConfig The cacheConfig that disables the block cache.
221    * @param current The current time.
222    * @throws IOException
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       // no need to clean, because the TTL is not set.
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       // no file found
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    * Gets the root dir of the mob files.
289    * It's {HBASE_DIR}/mobdir.
290    * @param conf The current configuration.
291    * @return the root dir of the mob file.
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    * Gets the qualified root dir of the mob files.
300    * @param conf The current configuration.
301    * @return The qualified root dir.
302    * @throws IOException
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    * Gets the region dir of the mob files.
313    * It's {HBASE_DIR}/mobdir/{namespace}/{tableName}/{regionEncodedName}.
314    * @param conf The current configuration.
315    * @param tableName The current table name.
316    * @return The region dir of the mob files.
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    * Gets the family dir of the mob files.
326    * It's {HBASE_DIR}/mobdir/{namespace}/{tableName}/{regionEncodedName}/{columnFamilyName}.
327    * @param conf The current configuration.
328    * @param tableName The current table name.
329    * @param familyName The current family name.
330    * @return The family dir of the mob files.
331    */
332   public static Path getMobFamilyPath(Configuration conf, TableName tableName, String familyName) {
333     return new Path(getMobRegionPath(conf, tableName), familyName);
334   }
335 
336   /**
337    * Gets the family dir of the mob files.
338    * It's {HBASE_DIR}/mobdir/{namespace}/{tableName}/{regionEncodedName}/{columnFamilyName}.
339    * @param regionPath The path of mob region which is a dummy one.
340    * @param familyName The current family name.
341    * @return The family dir of the mob files.
342    */
343   public static Path getMobFamilyPath(Path regionPath, String familyName) {
344     return new Path(regionPath, familyName);
345   }
346 
347   /**
348    * Gets the HRegionInfo of the mob files.
349    * This is a dummy region. The mob files are not saved in a region in HBase.
350    * This is only used in mob snapshot. It's internally used only.
351    * @param tableName
352    * @return A dummy mob region info.
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    * Gets whether the current HRegionInfo is a mob one.
362    * @param regionInfo The current HRegionInfo.
363    * @return If true, the current HRegionInfo is a mob one.
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    * Gets whether the current region name follows the pattern of a mob region name.
372    * @param tableName The current table name.
373    * @param regionName The current region name.
374    * @return True if the current region name follows the pattern of a mob region name.
375    */
376   public static boolean isMobRegionName(TableName tableName, byte[] regionName) {
377     return Bytes.equals(regionName, getMobRegionInfo(tableName).getRegionName());
378   }
379 
380   /**
381    * Gets the working directory of the mob compaction.
382    * @param root The root directory of the mob compaction.
383    * @param jobName The current job name.
384    * @return The directory of the mob compaction for the current job.
385    */
386   public static Path getCompactionWorkingPath(Path root, String jobName) {
387     return new Path(root, jobName);
388   }
389 
390   /**
391    * Archives the mob files.
392    * @param conf The current configuration.
393    * @param fs The current file system.
394    * @param tableName The table name.
395    * @param tableDir The table directory.
396    * @param family The name of the column family.
397    * @param storeFiles The files to be deleted.
398    * @throws IOException
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    * Creates a mob reference KeyValue.
408    * The value of the mob reference KeyValue is mobCellValueSize + mobFileName.
409    * @param cell The original Cell.
410    * @param fileName The mob file name where the mob reference KeyValue is written.
411    * @param tableNameTag The tag of the current table name. It's very important in
412    *                        cloning the snapshot.
413    * @return The mob reference KeyValue.
414    */
415   public static KeyValue createMobRefKeyValue(Cell cell, byte[] fileName, Tag tableNameTag) {
416     // Append the tags to the KeyValue.
417     // The key is same, the value is the filename of the mob file
418     List<Tag> tags = new ArrayList<Tag>();
419     // Add the ref tag as the 1st one.
420     tags.add(MobConstants.MOB_REF_TAG);
421     // Add the tag of the source table name, this table is where this mob file is flushed
422     // from.
423     // It's very useful in cloning the snapshot. When reading from the cloning table, we need to
424     // find the original mob files by this table name. For details please see cloning
425     // snapshot for mob files.
426     tags.add(tableNameTag);
427     // Add the existing tags.
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    * Creates a writer for the mob file in temp directory.
441    * @param conf The current configuration.
442    * @param fs The current file system.
443    * @param family The descriptor of the current column family.
444    * @param date The date string, its format is yyyymmmdd.
445    * @param basePath The basic path for a temp directory.
446    * @param maxKeyCount The key count.
447    * @param compression The compression algorithm.
448    * @param startKey The hex string of the start key.
449    * @param cacheConfig The current cache config.
450    * @return The writer for the mob file.
451    * @throws IOException
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    * Creates a writer for the ref file in temp directory.
465    * @param conf The current configuration.
466    * @param fs The current file system.
467    * @param family The descriptor of the current column family.
468    * @param basePath The basic path for a temp directory.
469    * @param maxKeyCount The key count.
470    * @param cacheConfig The current cache config.
471    * @return The writer for the mob file.
472    * @throws IOException
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    * Creates a writer for the mob file in temp directory.
491    * @param conf The current configuration.
492    * @param fs The current file system.
493    * @param family The descriptor of the current column family.
494    * @param date The date string, its format is yyyymmmdd.
495    * @param basePath The basic path for a temp directory.
496    * @param maxKeyCount The key count.
497    * @param compression The compression algorithm.
498    * @param startKey The start key.
499    * @param cacheConfig The current cache config.
500    * @return The writer for the mob file.
501    * @throws IOException
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    * Creates a writer for the del file in temp directory.
515    * @param conf The current configuration.
516    * @param fs The current file system.
517    * @param family The descriptor of the current column family.
518    * @param date The date string, its format is yyyymmmdd.
519    * @param basePath The basic path for a temp directory.
520    * @param maxKeyCount The key count.
521    * @param compression The compression algorithm.
522    * @param startKey The start key.
523    * @param cacheConfig The current cache config.
524    * @return The writer for the del file.
525    * @throws IOException
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    * Creates a writer for the del file in temp directory.
540    * @param conf The current configuration.
541    * @param fs The current file system.
542    * @param family The descriptor of the current column family.
543    * @param mobFileName The mob file name.
544    * @param basePath The basic path for a temp directory.
545    * @param maxKeyCount The key count.
546    * @param compression The compression algorithm.
547    * @param cacheConfig The current cache config.
548    * @return The writer for the mob file.
549    * @throws IOException
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    * Commits the mob file.
569    * @param conf The current configuration.
570    * @param fs The current file system.
571    * @param sourceFile The path where the mob file is saved.
572    * @param targetPath The directory path where the source file is renamed to.
573    * @param cacheConfig The current cache config.
574    * @return The target file path the source file is renamed to.
575    * @throws IOException
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    * Validates a mob file by opening and closing it.
598    * @param conf The current configuration.
599    * @param fs The current file system.
600    * @param path The path where the mob file is saved.
601    * @param cacheConfig The current cache config.
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    * Indicates whether the current mob ref cell has a valid value.
621    * A mob ref cell has a mob reference tag.
622    * The value of a mob ref cell consists of two parts, real mob value length and mob file name.
623    * The real mob value length takes 4 bytes.
624    * The remaining part is the mob file name.
625    * @param cell The mob ref cell.
626    * @return True if the cell has a valid value.
627    */
628   public static boolean hasValidMobRefCellValue(Cell cell) {
629     return cell.getValueLength() > Bytes.SIZEOF_INT;
630   }
631 
632   /**
633    * Gets the mob value length from the mob ref cell.
634    * A mob ref cell has a mob reference tag.
635    * The value of a mob ref cell consists of two parts, real mob value length and mob file name.
636    * The real mob value length takes 4 bytes.
637    * The remaining part is the mob file name.
638    * @param cell The mob ref cell.
639    * @return The real mob value length.
640    */
641   public static int getMobValueLength(Cell cell) {
642     return Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), Bytes.SIZEOF_INT);
643   }
644 
645   /**
646    * Gets the mob file name from the mob ref cell.
647    * A mob ref cell has a mob reference tag.
648    * The value of a mob ref cell consists of two parts, real mob value length and mob file name.
649    * The real mob value length takes 4 bytes.
650    * The remaining part is the mob file name.
651    * @param cell The mob ref cell.
652    * @return The mob file name.
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    * Gets the table name used in the table lock.
661    * The table lock name is a dummy one, it's not a table name. It's tableName + ".mobLock".
662    * @param tn The table name.
663    * @return The table name used in table lock.
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    * Performs the mob file compaction.
672    * @param conf the Configuration
673    * @param fs the file system
674    * @param tableName the table the compact
675    * @param hcd the column descriptor
676    * @param pool the thread pool
677    * @param tableLockManager the tableLock manager
678    * @param isForceAllFiles Whether add all mob files into the compaction.
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     // instantiate the mob file compactor.
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     // compact only for mob-enabled column.
695     // obtain a write table lock before performing compaction to avoid race condition
696     // with major compaction in mob-enabled column.
697     boolean tableLocked = false;
698     TableLock lock = null;
699     try {
700       // the tableLockManager might be null in testing. In that case, it is lock-free.
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    * Creates a thread pool.
725    * @param conf the Configuration
726    * @return A thread pool.
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             // waiting for a thread to pick up instead of throwing exceptions.
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    * Indicates whether return null value when the mob file is missing or corrupt.
753    * The information is set in the attribute "empty.value.on.mobcell.miss" of scan.
754    * @param scan The current scan.
755    * @return True if the readEmptyValueOnMobCellMiss is enabled.
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    * Archive mob store files
768    * @param conf The current configuration.
769    * @param fs The current file system.
770    * @param mobRegionInfo The mob family region info.
771    * @param mobFamilyDir The mob family directory.
772    * @param family The name of the column family.
773    * @throws IOException
774    */
775   public static void archiveMobStoreFiles(Configuration conf, FileSystem fs,
776       HRegionInfo mobRegionInfo, Path mobFamilyDir, byte[] family) throws IOException {
777     // disable the block cache.
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 }