View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.mob;
19  
20  import java.io.IOException;
21  import java.util.ArrayList;
22  import java.util.Date;
23  import java.util.List;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.classification.InterfaceAudience;
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.fs.Path;
30  import org.apache.hadoop.hbase.Cell;
31  import org.apache.hadoop.hbase.CellUtil;
32  import org.apache.hadoop.hbase.KeyValue;
33  import org.apache.hadoop.hbase.KeyValueUtil;
34  import org.apache.hadoop.hbase.Tag;
35  import org.apache.hadoop.hbase.TagType;
36  import org.apache.hadoop.hbase.client.Scan;
37  import org.apache.hadoop.hbase.regionserver.HMobStore;
38  import org.apache.hadoop.hbase.regionserver.HStore;
39  import org.apache.hadoop.hbase.regionserver.InternalScanner;
40  import org.apache.hadoop.hbase.regionserver.MobCompactionStoreScanner;
41  import org.apache.hadoop.hbase.regionserver.ScanType;
42  import org.apache.hadoop.hbase.regionserver.ScannerContext;
43  import org.apache.hadoop.hbase.regionserver.Store;
44  import org.apache.hadoop.hbase.regionserver.StoreFile;
45  import org.apache.hadoop.hbase.regionserver.StoreFile.Writer;
46  import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
47  import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
48  import org.apache.hadoop.hbase.regionserver.compactions.CompactionThroughputController;
49  import org.apache.hadoop.hbase.util.Bytes;
50  
51  /**
52   * Compact passed set of files in the mob-enabled column family.
53   */
54  @InterfaceAudience.Private
55  public class DefaultMobCompactor extends DefaultCompactor {
56  
57    private static final Log LOG = LogFactory.getLog(DefaultMobCompactor.class);
58    private long mobSizeThreshold;
59    private HMobStore mobStore;
60    public DefaultMobCompactor(Configuration conf, Store store) {
61      super(conf, store);
62      // The mob cells reside in the mob-enabled column family which is held by HMobStore.
63      // During the compaction, the compactor reads the cells from the mob files and
64      // probably creates new mob files. All of these operations are included in HMobStore,
65      // so we need to cast the Store to HMobStore.
66      if (!(store instanceof HMobStore)) {
67        throw new IllegalArgumentException("The store " + store + " is not a HMobStore");
68      }
69      mobStore = (HMobStore) store;
70      mobSizeThreshold = store.getFamily().getMobThreshold();
71    }
72  
73    /**
74     * Creates a writer for a new file in a temporary directory.
75     * @param fd The file details.
76     * @param shouldDropBehind Should the writer drop behind.
77     * @return Writer for a new StoreFile in the tmp dir.
78     * @throws IOException
79     */
80    @Override
81    protected Writer createTmpWriter(FileDetails fd, boolean shouldDropBehind) throws IOException {
82      // make this writer with tags always because of possible new cells with tags.
83      StoreFile.Writer writer = store.createWriterInTmp(fd.maxKeyCount, this.compactionCompression,
84        true, true, true, shouldDropBehind);
85      return writer;
86    }
87  
88    @Override
89    protected InternalScanner createScanner(Store store, List<StoreFileScanner> scanners,
90        ScanType scanType, long smallestReadPoint, long earliestPutTs) throws IOException {
91      Scan scan = new Scan();
92      scan.setMaxVersions(store.getFamily().getMaxVersions());
93      if (scanType == ScanType.COMPACT_DROP_DELETES) {
94        scanType = ScanType.COMPACT_RETAIN_DELETES;
95        return new MobCompactionStoreScanner(store, store.getScanInfo(), scan, scanners,
96            scanType, smallestReadPoint, earliestPutTs, true);
97      } else {
98        return new MobCompactionStoreScanner(store, store.getScanInfo(), scan, scanners,
99            scanType, smallestReadPoint, earliestPutTs, false);
100     }
101   }
102 
103   /**
104    * Performs compaction on a column family with the mob flag enabled.
105    * This is for when the mob threshold size has changed or if the mob
106    * column family mode has been toggled via an alter table statement.
107    * Compacts the files by the following rules.
108    * 1. If the cell has a mob reference tag, the cell's value is the path of the mob file.
109    * <ol>
110    * <li>
111    * If the value size of a cell is larger than the threshold, this cell is regarded as a mob,
112    * directly copy the (with mob tag) cell into the new store file.
113    * </li>
114    * <li>
115    * Otherwise, retrieve the mob cell from the mob file, and writes a copy of the cell into
116    * the new store file.
117    * </li>
118    * </ol>
119    * 2. If the cell doesn't have a reference tag.
120    * <ol>
121    * <li>
122    * If the value size of a cell is larger than the threshold, this cell is regarded as a mob,
123    * write this cell to a mob file, and write the path of this mob file to the store file.
124    * </li>
125    * <li>
126    * Otherwise, directly write this cell into the store file.
127    * </li>
128    * </ol>
129    * In the mob compaction, the {@link MobCompactionStoreScanner} is used as a scanner
130    * which could output the normal cells and delete markers together when required.
131    * After the major compaction on the normal hfiles, we have a guarantee that we have purged all
132    * deleted or old version mob refs, and the delete markers are written to a del file with the
133    * suffix _del. Because of this, it is safe to use the del file in the mob compaction.
134    * The mob compaction doesn't take place in the normal hfiles, it occurs directly in the
135    * mob files. When the small mob files are merged into bigger ones, the del file is added into
136    * the scanner to filter the deleted cells.
137    * @param fd File details
138    * @param scanner Where to read from.
139    * @param writer Where to write to.
140    * @param smallestReadPoint Smallest read point.
141    * @param cleanSeqId When true, remove seqId(used to be mvcc) value which is <= smallestReadPoint
142    * @param major Is a major compaction.
143    * @return Whether compaction ended; false if it was interrupted for any reason.
144    */
145   @Override
146   protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
147       long smallestReadPoint, boolean cleanSeqId,
148       CompactionThroughputController throughputController, boolean major) throws IOException {
149     if (!(scanner instanceof MobCompactionStoreScanner)) {
150       throw new IllegalArgumentException(
151           "The scanner should be an instance of MobCompactionStoreScanner");
152     }
153     MobCompactionStoreScanner compactionScanner = (MobCompactionStoreScanner) scanner;
154     int bytesWritten = 0;
155     // Since scanner.next() can return 'false' but still be delivering data,
156     // we have to use a do/while loop.
157     List<Cell> cells = new ArrayList<Cell>();
158     // Limit to "hbase.hstore.compaction.kv.max" (default 10) to avoid OOME
159     int closeCheckInterval = HStore.getCloseCheckInterval();
160     boolean hasMore;
161     Path path = MobUtils.getMobFamilyPath(conf, store.getTableName(), store.getColumnFamilyName());
162     byte[] fileName = null;
163     StoreFile.Writer mobFileWriter = null;
164     StoreFile.Writer delFileWriter = null;
165     long mobCells = 0;
166     long deleteMarkersCount = 0;
167     Tag tableNameTag = new Tag(TagType.MOB_TABLE_NAME_TAG_TYPE, store.getTableName()
168             .getName());
169     long mobCompactedIntoMobCellsCount = 0;
170     long mobCompactedFromMobCellsCount = 0;
171     long mobCompactedIntoMobCellsSize = 0;
172     long mobCompactedFromMobCellsSize = 0;
173     try {
174       try {
175         // If the mob file writer could not be created, directly write the cell to the store file.
176         mobFileWriter = mobStore.createWriterInTmp(new Date(fd.latestPutTs), fd.maxKeyCount,
177             store.getFamily().getCompression(), store.getRegionInfo().getStartKey());
178         fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
179       } catch (IOException e) {
180         LOG.error(
181             "Fail to create mob writer, "
182                 + "we will continue the compaction by writing MOB cells directly in store files",
183             e);
184       }
185       delFileWriter = mobStore.createDelFileWriterInTmp(new Date(fd.latestPutTs), fd.maxKeyCount,
186           store.getFamily().getCompression(), store.getRegionInfo().getStartKey());
187       ScannerContext scannerContext =
188         ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
189       do {
190         hasMore = compactionScanner.next(cells, scannerContext);
191         // output to writer:
192         for (Cell c : cells) {
193           // TODO remove the KeyValueUtil.ensureKeyValue before merging back to trunk.
194           KeyValue kv = KeyValueUtil.ensureKeyValue(c);
195           resetSeqId(smallestReadPoint, cleanSeqId, kv);
196           if (compactionScanner.isOutputDeleteMarkers() && CellUtil.isDelete(c)) {
197             delFileWriter.append(kv);
198             deleteMarkersCount++;
199           } else if (mobFileWriter == null || kv.getTypeByte() != KeyValue.Type.Put.getCode()) {
200             // If the mob file writer is null or the kv type is not put, directly write the cell
201             // to the store file.
202             writer.append(kv);
203           } else if (MobUtils.isMobReferenceCell(kv)) {
204             if (MobUtils.hasValidMobRefCellValue(kv)) {
205               int size = MobUtils.getMobValueLength(kv);
206               if (size > mobSizeThreshold) {
207                 // If the value size is larger than the threshold, it's regarded as a mob. Since
208                 // its value is already in the mob file, directly write this cell to the store file
209                 writer.append(kv);
210               } else {
211                 // If the value is not larger than the threshold, it's not regarded a mob. Retrieve
212                 // the mob cell from the mob file, and write it back to the store file.
213                 Cell cell = mobStore.resolve(kv, false);
214                 if (cell.getValueLength() != 0) {
215                   // put the mob data back to the store file
216                   KeyValue mobKv = KeyValueUtil.ensureKeyValue(cell);
217                   mobKv.setSequenceId(kv.getSequenceId());
218                   writer.append(mobKv);
219                   mobCompactedFromMobCellsCount++;
220                   mobCompactedFromMobCellsSize += cell.getValueLength();
221                 } else {
222                   // If the value of a file is empty, there might be issues when retrieving,
223                   // directly write the cell to the store file, and leave it to be handled by the
224                   // next compaction.
225                   writer.append(kv);
226                 }
227               }
228             } else {
229               LOG.warn("The value format of the KeyValue " + kv
230                   + " is wrong, its length is less than " + Bytes.SIZEOF_INT);
231               writer.append(kv);
232             }
233           } else if (kv.getValueLength() <= mobSizeThreshold) {
234             // If the value size of a cell is not larger than the threshold, directly write it to
235             // the store file.
236             writer.append(kv);
237           } else {
238             // If the value size of a cell is larger than the threshold, it's regarded as a mob,
239             // write this cell to a mob file, and write the path to the store file.
240             mobCells++;
241             // append the original keyValue in the mob file.
242             mobFileWriter.append(kv);
243             KeyValue reference = MobUtils.createMobRefKeyValue(kv, fileName, tableNameTag);
244             // write the cell whose value is the path of a mob file to the store file.
245             writer.append(reference);
246             mobCompactedIntoMobCellsCount++;
247             mobCompactedIntoMobCellsSize += kv.getValueLength();
248           }
249           ++progress.currentCompactedKVs;
250 
251           // check periodically to see if a system stop is requested
252           if (closeCheckInterval > 0) {
253             bytesWritten += kv.getLength();
254             if (bytesWritten > closeCheckInterval) {
255               bytesWritten = 0;
256               if (!store.areWritesEnabled()) {
257                 progress.cancel();
258                 return false;
259               }
260             }
261           }
262         }
263         cells.clear();
264       } while (hasMore);
265     } finally {
266       if (mobFileWriter != null) {
267         mobFileWriter.appendMetadata(fd.maxSeqId, major, mobCells);
268         mobFileWriter.close();
269       }
270       if (delFileWriter != null) {
271         delFileWriter.appendMetadata(fd.maxSeqId, major, deleteMarkersCount);
272         delFileWriter.close();
273       }
274     }
275     if (mobFileWriter != null) {
276       if (mobCells > 0) {
277         // If the mob file is not empty, commit it.
278         mobStore.commitFile(mobFileWriter.getPath(), path);
279       } else {
280         try {
281           // If the mob file is empty, delete it instead of committing.
282           store.getFileSystem().delete(mobFileWriter.getPath(), true);
283         } catch (IOException e) {
284           LOG.error("Fail to delete the temp mob file", e);
285         }
286       }
287     }
288     if (delFileWriter != null) {
289       if (deleteMarkersCount > 0) {
290         // If the del file is not empty, commit it.
291         // If the commit fails, the compaction is re-performed again.
292         mobStore.commitFile(delFileWriter.getPath(), path);
293       } else {
294         try {
295           // If the del file is empty, delete it instead of committing.
296           store.getFileSystem().delete(delFileWriter.getPath(), true);
297         } catch (IOException e) {
298           LOG.error("Fail to delete the temp del file", e);
299         }
300       }
301     }
302     mobStore.updateMobCompactedFromMobCellsCount(mobCompactedFromMobCellsCount);
303     mobStore.updateMobCompactedIntoMobCellsCount(mobCompactedIntoMobCellsCount);
304     mobStore.updateMobCompactedFromMobCellsSize(mobCompactedFromMobCellsSize);
305     mobStore.updateMobCompactedIntoMobCellsSize(mobCompactedIntoMobCellsSize);
306     progress.complete();
307     return true;
308   }
309 }