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.IOException;
22  import java.util.ArrayList;
23  import java.util.Date;
24  import java.util.List;
25  
26  import org.apache.commons.logging.Log;
27  import org.apache.commons.logging.LogFactory;
28  import org.apache.hadoop.classification.InterfaceAudience;
29  import org.apache.hadoop.conf.Configuration;
30  import org.apache.hadoop.fs.Path;
31  import org.apache.hadoop.hbase.Cell;
32  import org.apache.hadoop.hbase.HConstants;
33  import org.apache.hadoop.hbase.KeyValue;
34  import org.apache.hadoop.hbase.KeyValueUtil;
35  import org.apache.hadoop.hbase.Tag;
36  import org.apache.hadoop.hbase.TagType;
37  import org.apache.hadoop.hbase.monitoring.MonitoredTask;
38  import org.apache.hadoop.hbase.regionserver.DefaultStoreFlusher;
39  import org.apache.hadoop.hbase.regionserver.HMobStore;
40  import org.apache.hadoop.hbase.regionserver.InternalScanner;
41  import org.apache.hadoop.hbase.regionserver.MemStoreSnapshot;
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.util.Bytes;
46  
47  /**
48   * An implementation of the StoreFlusher. It extends the DefaultStoreFlusher.
49   * If the store is not a mob store, the flusher flushes the MemStore the same with
50   * DefaultStoreFlusher,
51   * If the store is a mob store, the flusher flushes the MemStore into two places.
52   * One is the store files of HBase, the other is the mob files.
53   * <ol>
54   * <li>Cells that are not PUT type or have the delete mark will be directly flushed to HBase.</li>
55   * <li>If the size of a cell value is larger than a threshold, it'll be flushed
56   * to a mob file, another cell with the path of this file will be flushed to HBase.</li>
57   * <li>If the size of a cell value is smaller than or equal with a threshold, it'll be flushed to
58   * HBase directly.</li>
59   * </ol>
60   *
61   */
62  @InterfaceAudience.Private
63  public class DefaultMobStoreFlusher extends DefaultStoreFlusher {
64  
65    private static final Log LOG = LogFactory.getLog(DefaultMobStoreFlusher.class);
66    private final Object flushLock = new Object();
67    private long mobCellValueSizeThreshold = 0;
68    private Path targetPath;
69    private HMobStore mobStore;
70  
71    public DefaultMobStoreFlusher(Configuration conf, Store store) throws IOException {
72      super(conf, store);
73      mobCellValueSizeThreshold = store.getFamily().getMobThreshold();
74      this.targetPath = MobUtils.getMobFamilyPath(conf, store.getTableName(),
75          store.getColumnFamilyName());
76      if (!this.store.getFileSystem().exists(targetPath)) {
77        this.store.getFileSystem().mkdirs(targetPath);
78      }
79      this.mobStore = (HMobStore) store;
80    }
81  
82    /**
83     * Flushes the snapshot of the MemStore.
84     * If this store is not a mob store, flush the cells in the snapshot to store files of HBase.
85     * If the store is a mob one, the flusher flushes the MemStore into two places.
86     * One is the store files of HBase, the other is the mob files.
87     * <ol>
88     * <li>Cells that are not PUT type or have the delete mark will be directly flushed to
89     * HBase.</li>
90     * <li>If the size of a cell value is larger than a threshold, it'll be
91     * flushed to a mob file, another cell with the path of this file will be flushed to HBase.</li>
92     * <li>If the size of a cell value is smaller than or equal with a threshold, it'll be flushed to
93     * HBase directly.</li>
94     * </ol>
95     */
96    @Override
97    public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushId,
98        MonitoredTask status) throws IOException {
99      ArrayList<Path> result = new ArrayList<Path>();
100     int cellsCount = snapshot.getCellsCount();
101     if (cellsCount == 0) return result; // don't flush if there are no entries
102 
103     // Use a store scanner to find which rows to flush.
104     long smallestReadPoint = store.getSmallestReadPoint();
105     InternalScanner scanner = createScanner(snapshot.getScanner(), smallestReadPoint);
106     if (scanner == null) {
107       return result; // NULL scanner returned from coprocessor hooks means skip normal processing
108     }
109     StoreFile.Writer writer;
110     try {
111       // TODO: We can fail in the below block before we complete adding this flush to
112       // list of store files. Add cleanup of anything put on filesystem if we fail.
113       synchronized (flushLock) {
114         status.setStatus("Flushing " + store + ": creating writer");
115         // Write the map out to the disk
116         writer = store.createWriterInTmp(cellsCount, store.getFamily().getCompression(),
117             false, true, true);
118         writer.setTimeRangeTracker(snapshot.getTimeRangeTracker());
119         try {
120           // It's a mob store, flush the cells in a mob way. This is the difference of flushing
121           // between a normal and a mob store.
122           performMobFlush(snapshot, cacheFlushId, scanner, writer, status);
123         } finally {
124           finalizeWriter(writer, cacheFlushId, status);
125         }
126       }
127     } finally {
128       scanner.close();
129     }
130     LOG.info("Flushed, sequenceid=" + cacheFlushId + ", memsize="
131         + snapshot.getSize() + ", hasBloomFilter=" + writer.hasGeneralBloom()
132         + ", into tmp file " + writer.getPath());
133     result.add(writer.getPath());
134     return result;
135   }
136 
137   /**
138    * Flushes the cells in the mob store.
139    * <ol>In the mob store, the cells with PUT type might have or have no mob tags.
140    * <li>If a cell does not have a mob tag, flushing the cell to different files depends
141    * on the value length. If the length is larger than a threshold, it's flushed to a
142    * mob file and the mob file is flushed to a store file in HBase. Otherwise, directly
143    * flush the cell to a store file in HBase.</li>
144    * <li>If a cell have a mob tag, its value is a mob file name, directly flush it
145    * to a store file in HBase.</li>
146    * </ol>
147    * @param snapshot Memstore snapshot.
148    * @param cacheFlushId Log cache flush sequence number.
149    * @param scanner The scanner of memstore snapshot.
150    * @param writer The store file writer.
151    * @param status Task that represents the flush operation and may be updated with status.
152    * @throws IOException
153    */
154   protected void performMobFlush(MemStoreSnapshot snapshot, long cacheFlushId,
155       InternalScanner scanner, StoreFile.Writer writer, MonitoredTask status) throws IOException {
156     StoreFile.Writer mobFileWriter = null;
157     int compactionKVMax = conf.getInt(HConstants.COMPACTION_KV_MAX,
158         HConstants.COMPACTION_KV_MAX_DEFAULT);
159     long mobCount = 0;
160     long mobSize = 0;
161     long time = snapshot.getTimeRangeTracker().getMaximumTimestamp();
162     mobFileWriter = mobStore.createWriterInTmp(new Date(time), snapshot.getCellsCount(),
163         store.getFamily().getCompression(), store.getRegionInfo().getStartKey());
164     // the target path is {tableName}/.mob/{cfName}/mobFiles
165     // the relative path is mobFiles
166     byte[] fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
167     try {
168       Tag tableNameTag = new Tag(TagType.MOB_TABLE_NAME_TAG_TYPE, store.getTableName()
169           .getName());
170       List<Cell> cells = new ArrayList<Cell>();
171       boolean hasMore;
172       ScannerContext scannerContext =
173               ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
174       
175       do {
176         hasMore = scanner.next(cells, scannerContext);
177         if (!cells.isEmpty()) {
178           for (Cell c : cells) {
179             // If we know that this KV is going to be included always, then let us
180             // set its memstoreTS to 0. This will help us save space when writing to
181             // disk.
182             KeyValue kv = KeyValueUtil.ensureKeyValue(c);
183             if (kv.getValueLength() <= mobCellValueSizeThreshold || MobUtils.isMobReferenceCell(kv)
184                 || kv.getTypeByte() != KeyValue.Type.Put.getCode()) {
185               writer.append(kv);
186             } else {
187               // append the original keyValue in the mob file.
188               mobFileWriter.append(kv);
189               mobSize += kv.getValueLength();
190               mobCount++;
191 
192               // append the tags to the KeyValue.
193               // The key is same, the value is the filename of the mob file
194               KeyValue reference = MobUtils.createMobRefKeyValue(kv, fileName, tableNameTag);
195               writer.append(reference);
196             }
197           }
198           cells.clear();
199         }
200       } while (hasMore);
201     } finally {
202       status.setStatus("Flushing mob file " + store + ": appending metadata");
203       mobFileWriter.appendMetadata(cacheFlushId, false, mobCount);
204       status.setStatus("Flushing mob file " + store + ": closing flushed file");
205       mobFileWriter.close();
206     }
207 
208     if (mobCount > 0) {
209       // commit the mob file from temp folder to target folder.
210       // If the mob file is committed successfully but the store file is not,
211       // the committed mob file will be handled by the sweep tool as an unused
212       // file.
213       mobStore.commitFile(mobFileWriter.getPath(), targetPath);
214       mobStore.updateMobFlushCount();
215       mobStore.updateMobFlushedCellsCount(mobCount);
216       mobStore.updateMobFlushedCellsSize(mobSize);
217     } else {
218       try {
219         // If the mob file is empty, delete it instead of committing.
220         store.getFileSystem().delete(mobFileWriter.getPath(), true);
221       } catch (IOException e) {
222         LOG.error("Fail to delete the temp mob file", e);
223       }
224     }
225   }
226 }