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.filecompactions;
20  
21  import java.io.FileNotFoundException;
22  import java.io.IOException;
23  import java.util.ArrayList;
24  import java.util.Collection;
25  import java.util.Collections;
26  import java.util.Date;
27  import java.util.HashMap;
28  import java.util.List;
29  import java.util.Map;
30  import java.util.Map.Entry;
31  import java.util.concurrent.Callable;
32  import java.util.concurrent.ExecutorService;
33  import java.util.concurrent.Future;
34  
35  import org.apache.commons.logging.Log;
36  import org.apache.commons.logging.LogFactory;
37  import org.apache.hadoop.classification.InterfaceAudience;
38  import org.apache.hadoop.conf.Configuration;
39  import org.apache.hadoop.fs.FileStatus;
40  import org.apache.hadoop.fs.FileSystem;
41  import org.apache.hadoop.fs.Path;
42  import org.apache.hadoop.hbase.Cell;
43  import org.apache.hadoop.hbase.HColumnDescriptor;
44  import org.apache.hadoop.hbase.HConstants;
45  import org.apache.hadoop.hbase.KeyValue;
46  import org.apache.hadoop.hbase.KeyValueUtil;
47  import org.apache.hadoop.hbase.TableName;
48  import org.apache.hadoop.hbase.Tag;
49  import org.apache.hadoop.hbase.TagType;
50  import org.apache.hadoop.hbase.client.HTable;
51  import org.apache.hadoop.hbase.client.Scan;
52  import org.apache.hadoop.hbase.io.HFileLink;
53  import org.apache.hadoop.hbase.io.hfile.CacheConfig;
54  import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
55  import org.apache.hadoop.hbase.mob.MobConstants;
56  import org.apache.hadoop.hbase.mob.MobFileName;
57  import org.apache.hadoop.hbase.mob.MobUtils;
58  import org.apache.hadoop.hbase.mob.filecompactions.MobFileCompactionRequest.CompactionType;
59  import org.apache.hadoop.hbase.mob.filecompactions.PartitionedMobFileCompactionRequest.CompactionPartition;
60  import org.apache.hadoop.hbase.mob.filecompactions.PartitionedMobFileCompactionRequest.CompactionPartitionId;
61  import org.apache.hadoop.hbase.regionserver.*;
62  import org.apache.hadoop.hbase.regionserver.StoreFile.Writer;
63  import org.apache.hadoop.hbase.regionserver.ScannerContext;
64  import org.apache.hadoop.hbase.util.Bytes;
65  import org.apache.hadoop.hbase.util.Pair;
66  
67  /**
68   * An implementation of {@link MobFileCompactor} that compacts the mob files in partitions.
69   */
70  @InterfaceAudience.Private
71  public class PartitionedMobFileCompactor extends MobFileCompactor {
72  
73    private static final Log LOG = LogFactory.getLog(PartitionedMobFileCompactor.class);
74    protected long mergeableSize;
75    protected int delFileMaxCount;
76    /** The number of files compacted in a batch */
77    protected int compactionBatchSize;
78    protected int compactionKVMax;
79  
80    private Path tempPath;
81    private Path bulkloadPath;
82    private CacheConfig compactionCacheConfig;
83    private Tag tableNameTag;
84  
85    public PartitionedMobFileCompactor(Configuration conf, FileSystem fs, TableName tableName,
86      HColumnDescriptor column, ExecutorService pool) {
87      super(conf, fs, tableName, column, pool);
88      mergeableSize = conf.getLong(MobConstants.MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD,
89        MobConstants.DEFAULT_MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD);
90      delFileMaxCount = conf.getInt(MobConstants.MOB_DELFILE_MAX_COUNT,
91        MobConstants.DEFAULT_MOB_DELFILE_MAX_COUNT);
92      // default is 100
93      compactionBatchSize = conf.getInt(MobConstants.MOB_FILE_COMPACTION_BATCH_SIZE,
94        MobConstants.DEFAULT_MOB_FILE_COMPACTION_BATCH_SIZE);
95      tempPath = new Path(MobUtils.getMobHome(conf), MobConstants.TEMP_DIR_NAME);
96      bulkloadPath = new Path(tempPath, new Path(MobConstants.BULKLOAD_DIR_NAME, new Path(
97        tableName.getNamespaceAsString(), tableName.getQualifierAsString())));
98      compactionKVMax = this.conf.getInt(HConstants.COMPACTION_KV_MAX,
99        HConstants.COMPACTION_KV_MAX_DEFAULT);
100     Configuration copyOfConf = new Configuration(conf);
101     copyOfConf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0f);
102     compactionCacheConfig = new CacheConfig(copyOfConf);
103     tableNameTag = new Tag(TagType.MOB_TABLE_NAME_TAG_TYPE, tableName.getName());
104   }
105 
106   @Override
107   public List<Path> compact(List<FileStatus> files, boolean isForceAllFiles) throws IOException {
108     if (files == null || files.isEmpty()) {
109       LOG.info("No candidate mob files");
110       return null;
111     }
112     LOG.info("isForceAllFiles: " + isForceAllFiles);
113     // find the files to compact.
114     PartitionedMobFileCompactionRequest request = select(files, isForceAllFiles);
115     // compact the files.
116     return performCompaction(request);
117   }
118 
119   /**
120    * Selects the compacted mob/del files.
121    * Iterates the candidates to find out all the del files and small mob files.
122    * @param candidates All the candidates.
123    * @param isForceAllFiles Whether add all mob files into the compaction.
124    * @return A compaction request.
125    * @throws IOException
126    */
127   protected PartitionedMobFileCompactionRequest select(List<FileStatus> candidates,
128     boolean isForceAllFiles) throws IOException {
129     Collection<FileStatus> allDelFiles = new ArrayList<FileStatus>();
130     Map<CompactionPartitionId, CompactionPartition> filesToCompact =
131       new HashMap<CompactionPartitionId, CompactionPartition>();
132     int selectedFileCount = 0;
133     int irrelevantFileCount = 0;
134     for (FileStatus file : candidates) {
135       if (!file.isFile()) {
136         irrelevantFileCount++;
137         continue;
138       }
139       // group the del files and small files.
140       FileStatus linkedFile = file;
141       if (HFileLink.isHFileLink(file.getPath())) {
142         HFileLink link = HFileLink.buildFromHFileLinkPattern(conf, file.getPath());
143         linkedFile = getLinkedFileStatus(link);
144         if (linkedFile == null) {
145           // If the linked file cannot be found, regard it as an irrelevantFileCount file
146           irrelevantFileCount++;
147           continue;
148         }
149       }
150       if (StoreFileInfo.isDelFile(linkedFile.getPath())) {
151         allDelFiles.add(file);
152       } else if (isForceAllFiles || linkedFile.getLen() < mergeableSize) {
153         // add all files if isForceAllFiles is true,
154         // otherwise add the small files to the merge pool
155         MobFileName fileName = MobFileName.create(linkedFile.getPath().getName());
156         CompactionPartitionId id = new CompactionPartitionId(fileName.getStartKey(),
157           fileName.getDate());
158         CompactionPartition compactionPartition = filesToCompact.get(id);
159         if (compactionPartition == null) {
160           compactionPartition = new CompactionPartition(id);
161           compactionPartition.addFile(file);
162           filesToCompact.put(id, compactionPartition);
163         } else {
164           compactionPartition.addFile(file);
165         }
166         selectedFileCount++;
167       }
168     }
169     PartitionedMobFileCompactionRequest request = new PartitionedMobFileCompactionRequest(
170       filesToCompact.values(), allDelFiles);
171     if (candidates.size() == (allDelFiles.size() + selectedFileCount + irrelevantFileCount)) {
172       // all the files are selected
173       request.setCompactionType(CompactionType.ALL_FILES);
174     }
175     LOG.info("The compaction type is " + request.getCompactionType() + ", the request has "
176       + allDelFiles.size() + " del files, " + selectedFileCount + " selected files, and "
177       + irrelevantFileCount + " irrelevant files");
178     return request;
179   }
180 
181   /**
182    * Performs the compaction on the selected files.
183    * <ol>
184    * <li>Compacts the del files.</li>
185    * <li>Compacts the selected small mob files and all the del files.</li>
186    * <li>If all the candidates are selected, delete the del files.</li>
187    * </ol>
188    * @param request The compaction request.
189    * @return The paths of new mob files generated in the compaction.
190    * @throws IOException
191    */
192   protected List<Path> performCompaction(PartitionedMobFileCompactionRequest request)
193     throws IOException {
194     // merge the del files
195     List<Path> delFilePaths = new ArrayList<Path>();
196     for (FileStatus delFile : request.delFiles) {
197       delFilePaths.add(delFile.getPath());
198     }
199     List<Path> newDelPaths = compactDelFiles(request, delFilePaths);
200     List<StoreFile> newDelFiles = new ArrayList<StoreFile>();
201     List<Path> paths = null;
202     try {
203       for (Path newDelPath : newDelPaths) {
204         StoreFile sf = new StoreFile(fs, newDelPath, conf, compactionCacheConfig, BloomType.NONE);
205         // pre-create reader of a del file to avoid race condition when opening the reader in each
206         // partition.
207         sf.createReader();
208         newDelFiles.add(sf);
209       }
210       LOG.info("After merging, there are " + newDelFiles.size() + " del files");
211       // compact the mob files by partitions.
212       paths = compactMobFiles(request, newDelFiles);
213       LOG.info("After compaction, there are " + paths.size() + " mob files");
214     } finally {
215       closeStoreFileReaders(newDelFiles);
216     }
217     // archive the del files if all the mob files are selected.
218     if (request.type == CompactionType.ALL_FILES && !newDelPaths.isEmpty()) {
219       LOG.info("After a mob file compaction with all files selected, archiving the del files "
220         + newDelFiles);
221       try {
222         MobUtils.removeMobFiles(conf, fs, tableName, mobTableDir, column.getName(), newDelFiles);
223       } catch (IOException e) {
224         LOG.error("Failed to archive the del files " + newDelFiles, e);
225       }
226     }
227     return paths;
228   }
229 
230   /**
231    * Compacts the selected small mob files and all the del files.
232    * @param request The compaction request.
233    * @param delFiles The del files.
234    * @return The paths of new mob files after compactions.
235    * @throws IOException
236    */
237   protected List<Path> compactMobFiles(final PartitionedMobFileCompactionRequest request,
238     final List<StoreFile> delFiles) throws IOException {
239     Collection<CompactionPartition> partitions = request.compactionPartitions;
240     if (partitions == null || partitions.isEmpty()) {
241       LOG.info("No partitions of mob files");
242       return Collections.emptyList();
243     }
244     List<Path> paths = new ArrayList<Path>();
245     final HTable table = new HTable(conf, tableName);
246     try {
247       Map<CompactionPartitionId, Future<List<Path>>> results =
248         new HashMap<CompactionPartitionId, Future<List<Path>>>();
249       // compact the mob files by partitions in parallel.
250       for (final CompactionPartition partition : partitions) {
251         results.put(partition.getPartitionId(), pool.submit(new Callable<List<Path>>() {
252           @Override
253           public List<Path> call() throws Exception {
254             LOG.info("Compacting mob files for partition " + partition.getPartitionId());
255             return compactMobFilePartition(request, partition, delFiles, table);
256           }
257         }));
258       }
259       // compact the partitions in parallel.
260       boolean hasFailure = false;
261       for (Entry<CompactionPartitionId, Future<List<Path>>> result : results.entrySet()) {
262         try {
263           paths.addAll(result.getValue().get());
264         } catch (Exception e) {
265           // just log the error
266           LOG.error("Failed to compact the partition " + result.getKey(), e);
267           hasFailure = true;
268         }
269       }
270       if (hasFailure) {
271         // if any partition fails in the compaction, directly throw an exception.
272         throw new IOException("Failed to compact the partitions");
273       }
274     } finally {
275       try {
276         table.close();
277       } catch (IOException e) {
278         LOG.error("Failed to close the HTable", e);
279       }
280     }
281     return paths;
282   }
283 
284   /**
285    * Compacts a partition of selected small mob files and all the del files.
286    * @param request The compaction request.
287    * @param partition A compaction partition.
288    * @param delFiles The del files.
289    * @param table The current table.
290    * @return The paths of new mob files after compactions.
291    * @throws IOException
292    */
293   private List<Path> compactMobFilePartition(PartitionedMobFileCompactionRequest request,
294     CompactionPartition partition, List<StoreFile> delFiles, HTable table) throws IOException {
295     List<Path> newFiles = new ArrayList<Path>();
296     List<FileStatus> files = partition.listFiles();
297     int offset = 0;
298     Path bulkloadPathOfPartition = new Path(bulkloadPath, partition.getPartitionId().toString());
299     Path bulkloadColumnPath = new Path(bulkloadPathOfPartition, column.getNameAsString());
300     while (offset < files.size()) {
301       int batch = compactionBatchSize;
302       if (files.size() - offset < compactionBatchSize) {
303         batch = files.size() - offset;
304       }
305       if (batch == 1 && delFiles.isEmpty()) {
306         // only one file left and no del files, do not compact it,
307         // and directly add it to the new files.
308         newFiles.add(files.get(offset).getPath());
309         offset++;
310         continue;
311       }
312       // clean the bulkload directory to avoid loading old files.
313       fs.delete(bulkloadPathOfPartition, true);
314       // add the selected mob files and del files into filesToCompact
315       List<StoreFile> filesToCompact = new ArrayList<StoreFile>();
316       for (int i = offset; i < batch + offset; i++) {
317         StoreFile sf = new StoreFile(fs, files.get(i).getPath(), conf, compactionCacheConfig,
318           BloomType.NONE);
319         filesToCompact.add(sf);
320       }
321       filesToCompact.addAll(delFiles);
322       // compact the mob files in a batch.
323       compactMobFilesInBatch(request, partition, table, filesToCompact, batch,
324         bulkloadPathOfPartition, bulkloadColumnPath, newFiles);
325       // move to the next batch.
326       offset += batch;
327     }
328     LOG.info("Compaction is finished. The number of mob files is changed from " + files.size()
329       + " to " + newFiles.size());
330     return newFiles;
331   }
332 
333   /**
334    * Closes the readers of store files.
335    * @param storeFiles The store files to be closed.
336    */
337   private void closeStoreFileReaders(List<StoreFile> storeFiles) {
338     for (StoreFile storeFile : storeFiles) {
339       try {
340         storeFile.closeReader(true);
341       } catch (IOException e) {
342         LOG.warn("Failed to close the reader on store file " + storeFile.getPath(), e);
343       }
344     }
345   }
346 
347   /**
348    * Compacts a partition of selected small mob files and all the del files in a batch.
349    * @param request The compaction request.
350    * @param partition A compaction partition.
351    * @param table The current table.
352    * @param filesToCompact The files to be compacted.
353    * @param batch The number of mob files to be compacted in a batch.
354    * @param bulkloadPathOfPartition The directory where the bulkload column of the current
355    *        partition is saved.
356    * @param bulkloadColumnPath The directory where the bulkload files of current partition
357    *        are saved.
358    * @param newFiles The paths of new mob files after compactions.
359    * @throws IOException
360    */
361   private void compactMobFilesInBatch(PartitionedMobFileCompactionRequest request,
362     CompactionPartition partition, HTable table, List<StoreFile> filesToCompact, int batch,
363     Path bulkloadPathOfPartition, Path bulkloadColumnPath, List<Path> newFiles)
364     throws IOException {
365     // open scanner to the selected mob files and del files.
366     StoreScanner scanner = createScanner(filesToCompact, ScanType.COMPACT_DROP_DELETES);
367     // the mob files to be compacted, not include the del files.
368     List<StoreFile> mobFilesToCompact = filesToCompact.subList(0, batch);
369     // Pair(maxSeqId, cellsCount)
370     Pair<Long, Long> fileInfo = getFileInfo(mobFilesToCompact);
371     // open writers for the mob files and new ref store files.
372     Writer writer = null;
373     Writer refFileWriter = null;
374     Path filePath = null;
375     Path refFilePath = null;
376     long mobCells = 0;
377     try {
378       writer = MobUtils.createWriter(conf, fs, column, partition.getPartitionId().getDate(),
379         tempPath, Long.MAX_VALUE, column.getCompactionCompression(), partition.getPartitionId()
380           .getStartKey(), compactionCacheConfig);
381       filePath = writer.getPath();
382       byte[] fileName = Bytes.toBytes(filePath.getName());
383       // create a temp file and open a writer for it in the bulkloadPath
384       refFileWriter = MobUtils.createRefFileWriter(conf, fs, column, bulkloadColumnPath, fileInfo
385         .getSecond().longValue(), compactionCacheConfig);
386       refFilePath = refFileWriter.getPath();
387       List<Cell> cells = new ArrayList<Cell>();
388       boolean hasMore = false;
389 
390       ScannerContext scannerContext =
391           ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
392       do {
393         hasMore = scanner.next(cells, scannerContext);
394         for (Cell cell : cells) {
395           // TODO remove this after the new code are introduced.
396           KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
397           // write the mob cell to the mob file.
398           writer.append(kv);
399           // write the new reference cell to the store file.
400           KeyValue reference = MobUtils.createMobRefKeyValue(kv, fileName, tableNameTag);
401           refFileWriter.append(reference);
402           mobCells++;
403         }
404         cells.clear();
405       } while (hasMore);
406     } finally {
407       // close the scanner.
408       scanner.close();
409       // append metadata to the mob file, and close the mob file writer.
410       closeMobFileWriter(writer, fileInfo.getFirst(), mobCells);
411       // append metadata and bulkload info to the ref mob file, and close the writer.
412       closeRefFileWriter(refFileWriter, fileInfo.getFirst(), request.selectionTime);
413     }
414     if (mobCells > 0) {
415       // commit mob file
416       MobUtils.commitFile(conf, fs, filePath, mobFamilyDir, compactionCacheConfig);
417       // bulkload the ref file
418       bulkloadRefFile(table, bulkloadPathOfPartition, filePath.getName());
419       newFiles.add(new Path(mobFamilyDir, filePath.getName()));
420     } else {
421       // remove the new files
422       // the mob file is empty, delete it instead of committing.
423       deletePath(filePath);
424       // the ref file is empty, delete it instead of committing.
425       deletePath(refFilePath);
426     }
427     // archive the old mob files, do not archive the del files.
428     try {
429       closeStoreFileReaders(mobFilesToCompact);
430       MobUtils
431         .removeMobFiles(conf, fs, tableName, mobTableDir, column.getName(), mobFilesToCompact);
432     } catch (IOException e) {
433       LOG.error("Failed to archive the files " + mobFilesToCompact, e);
434     }
435   }
436 
437   /**
438    * Compacts the del files in batches which avoids opening too many files.
439    * @param request The compaction request.
440    * @param delFilePaths
441    * @return The paths of new del files after merging or the original files if no merging
442    *         is necessary.
443    * @throws IOException
444    */
445   protected List<Path> compactDelFiles(PartitionedMobFileCompactionRequest request,
446     List<Path> delFilePaths) throws IOException {
447     if (delFilePaths.size() <= delFileMaxCount) {
448       return delFilePaths;
449     }
450     // when there are more del files than the number that is allowed, merge it firstly.
451     int offset = 0;
452     List<Path> paths = new ArrayList<Path>();
453     while (offset < delFilePaths.size()) {
454       // get the batch
455       int batch = compactionBatchSize;
456       if (delFilePaths.size() - offset < compactionBatchSize) {
457         batch = delFilePaths.size() - offset;
458       }
459       List<StoreFile> batchedDelFiles = new ArrayList<StoreFile>();
460       if (batch == 1) {
461         // only one file left, do not compact it, directly add it to the new files.
462         paths.add(delFilePaths.get(offset));
463         offset++;
464         continue;
465       }
466       for (int i = offset; i < batch + offset; i++) {
467         batchedDelFiles.add(new StoreFile(fs, delFilePaths.get(i), conf, compactionCacheConfig,
468           BloomType.NONE));
469       }
470       // compact the del files in a batch.
471       paths.add(compactDelFilesInBatch(request, batchedDelFiles));
472       // move to the next batch.
473       offset += batch;
474     }
475     return compactDelFiles(request, paths);
476   }
477 
478   /**
479    * Compacts the del file in a batch.
480    * @param request The compaction request.
481    * @param delFiles The del files.
482    * @return The path of new del file after merging.
483    * @throws IOException
484    */
485   private Path compactDelFilesInBatch(PartitionedMobFileCompactionRequest request,
486     List<StoreFile> delFiles) throws IOException {
487     // create a scanner for the del files.
488     StoreScanner scanner = createScanner(delFiles, ScanType.COMPACT_RETAIN_DELETES);
489     Writer writer = null;
490     Path filePath = null;
491     try {
492       writer = MobUtils.createDelFileWriter(conf, fs, column,
493         MobUtils.formatDate(new Date(request.selectionTime)), tempPath, Long.MAX_VALUE,
494         column.getCompactionCompression(), HConstants.EMPTY_START_ROW, compactionCacheConfig);
495       filePath = writer.getPath();
496       List<Cell> cells = new ArrayList<Cell>();
497       boolean hasMore = false;
498       ScannerContext scannerContext =
499           ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
500       do {
501         hasMore = scanner.next(cells, scannerContext);
502         for (Cell cell : cells) {
503           // TODO remove this after the new code are introduced.
504           KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
505           writer.append(kv);
506         }
507         cells.clear();
508       } while (hasMore);
509     } finally {
510       scanner.close();
511       if (writer != null) {
512         try {
513           writer.close();
514         } catch (IOException e) {
515           LOG.error("Failed to close the writer of the file " + filePath, e);
516         }
517       }
518     }
519     // commit the new del file
520     Path path = MobUtils.commitFile(conf, fs, filePath, mobFamilyDir, compactionCacheConfig);
521     // archive the old del files
522     try {
523       MobUtils.removeMobFiles(conf, fs, tableName, mobTableDir, column.getName(), delFiles);
524     } catch (IOException e) {
525       LOG.error("Failed to archive the old del files " + delFiles, e);
526     }
527     return path;
528   }
529 
530   /**
531    * Creates a store scanner.
532    * @param filesToCompact The files to be compacted.
533    * @param scanType The scan type.
534    * @return The store scanner.
535    * @throws IOException
536    */
537   private StoreScanner createScanner(List<StoreFile> filesToCompact, ScanType scanType)
538     throws IOException {
539     List scanners = StoreFileScanner.getScannersForStoreFiles(filesToCompact, false, true, false,
540       false, HConstants.LATEST_TIMESTAMP);
541     Scan scan = new Scan();
542     scan.setMaxVersions(column.getMaxVersions());
543     long ttl = HStore.determineTTLFromFamily(column);
544     ScanInfo scanInfo = new ScanInfo(conf, column, ttl, 0, KeyValue.COMPARATOR);
545     StoreScanner scanner = new StoreScanner(scan, scanInfo, scanType, null, scanners, 0L,
546       HConstants.LATEST_TIMESTAMP);
547     return scanner;
548   }
549 
550   /**
551    * Bulkloads the current file.
552    * @param table The current table.
553    * @param bulkloadDirectory The path of bulkload directory.
554    * @param fileName The current file name.
555    * @throws IOException
556    */
557   private void bulkloadRefFile(HTable table, Path bulkloadDirectory, String fileName)
558     throws IOException {
559     // bulkload the ref file
560     try {
561       LoadIncrementalHFiles bulkload = new LoadIncrementalHFiles(conf);
562       bulkload.doBulkLoad(bulkloadDirectory, table);
563     } catch (Exception e) {
564       // delete the committed mob file
565       deletePath(new Path(mobFamilyDir, fileName));
566       throw new IOException(e);
567     } finally {
568       // delete the bulkload files in bulkloadPath
569       deletePath(bulkloadDirectory);
570     }
571   }
572 
573   /**
574    * Closes the mob file writer.
575    * @param writer The mob file writer.
576    * @param maxSeqId Maximum sequence id.
577    * @param mobCellsCount The number of mob cells.
578    * @throws IOException
579    */
580   private void closeMobFileWriter(Writer writer, long maxSeqId, long mobCellsCount)
581     throws IOException {
582     if (writer != null) {
583       writer.appendMetadata(maxSeqId, false, mobCellsCount);
584       try {
585         writer.close();
586       } catch (IOException e) {
587         LOG.error("Failed to close the writer of the file " + writer.getPath(), e);
588       }
589     }
590   }
591 
592   /**
593    * Closes the ref file writer.
594    * @param writer The ref file writer.
595    * @param maxSeqId Maximum sequence id.
596    * @param bulkloadTime The timestamp at which the bulk load file is created.
597    * @throws IOException
598    */
599   private void closeRefFileWriter(Writer writer, long maxSeqId, long bulkloadTime)
600     throws IOException {
601     if (writer != null) {
602       writer.appendMetadata(maxSeqId, false);
603       writer.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(bulkloadTime));
604       try {
605         writer.close();
606       } catch (IOException e) {
607         LOG.error("Failed to close the writer of the ref file " + writer.getPath(), e);
608       }
609     }
610   }
611 
612   /**
613    * Gets the max seqId and number of cells of the store files.
614    * @param storeFiles The store files.
615    * @return The pair of the max seqId and number of cells of the store files.
616    * @throws IOException
617    */
618   private Pair<Long, Long> getFileInfo(List<StoreFile> storeFiles) throws IOException {
619     long maxSeqId = 0;
620     long maxKeyCount = 0;
621     for (StoreFile sf : storeFiles) {
622       // the readers will be closed later after the merge.
623       maxSeqId = Math.max(maxSeqId, sf.getMaxSequenceId());
624       byte[] count = sf.createReader().loadFileInfo().get(StoreFile.MOB_CELLS_COUNT);
625       if (count != null) {
626         maxKeyCount += Bytes.toLong(count);
627       }
628     }
629     return new Pair<Long, Long>(Long.valueOf(maxSeqId), Long.valueOf(maxKeyCount));
630   }
631 
632   /**
633    * Deletes a file.
634    * @param path The path of the file to be deleted.
635    */
636   private void deletePath(Path path) {
637     try {
638       if (path != null) {
639         fs.delete(path, true);
640       }
641     } catch (IOException e) {
642       LOG.error("Failed to delete the file " + path, e);
643     }
644   }
645 
646   private FileStatus getLinkedFileStatus(HFileLink link) throws IOException {
647     Path[] locations = link.getLocations();
648     for (Path location : locations) {
649       FileStatus file = getFileStatus(location);
650       if (file != null) {
651         return file;
652       }
653     }
654     return null;
655   }
656 
657   private FileStatus getFileStatus(Path path) throws IOException {
658     try {
659       if (path != null) {
660         FileStatus file = fs.getFileStatus(path);
661         return file;
662       }
663     } catch (FileNotFoundException e) {
664       LOG.warn("The file " + path + " can not be found", e);
665     }
666     return null;
667   }
668 }