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.regionserver.compactions;
19  
20  import java.io.IOException;
21  import java.io.InterruptedIOException;
22  import java.security.PrivilegedExceptionAction;
23  import java.util.ArrayList;
24  import java.util.Collection;
25  import java.util.List;
26  import java.util.Map;
27  import java.util.concurrent.atomic.AtomicInteger;
28  
29  import org.apache.commons.logging.Log;
30  import org.apache.commons.logging.LogFactory;
31  import org.apache.hadoop.conf.Configuration;
32  import org.apache.hadoop.hbase.Cell;
33  import org.apache.hadoop.hbase.CellUtil;
34  import org.apache.hadoop.hbase.HConstants;
35  import org.apache.hadoop.hbase.KeyValueUtil;
36  import org.apache.hadoop.hbase.classification.InterfaceAudience;
37  import org.apache.hadoop.hbase.client.Scan;
38  import org.apache.hadoop.hbase.io.compress.Compression;
39  import org.apache.hadoop.hbase.io.hfile.HFile.FileInfo;
40  import org.apache.hadoop.hbase.io.hfile.HFileWriterV2;
41  import org.apache.hadoop.hbase.regionserver.HStore;
42  import org.apache.hadoop.hbase.regionserver.InternalScanner;
43  import org.apache.hadoop.hbase.regionserver.ScanType;
44  import org.apache.hadoop.hbase.regionserver.ScannerContext;
45  import org.apache.hadoop.hbase.regionserver.Store;
46  import org.apache.hadoop.hbase.regionserver.StoreFile;
47  import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
48  import org.apache.hadoop.hbase.regionserver.StoreScanner;
49  import org.apache.hadoop.hbase.security.User;
50  import org.apache.hadoop.hbase.util.Bytes;
51  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
52  import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
53  
54  /**
55   * A compactor is a compaction algorithm associated a given policy. Base class also contains
56   * reusable parts for implementing compactors (what is common and what isn't is evolving).
57   */
58  @InterfaceAudience.Private
59  public abstract class Compactor {
60    private static final Log LOG = LogFactory.getLog(Compactor.class);
61    protected CompactionProgress progress;
62    protected Configuration conf;
63    protected Store store;
64  
65    private int compactionKVMax;
66    protected Compression.Algorithm compactionCompression;
67    
68    /** specify how many days to keep MVCC values during major compaction **/ 
69    protected int keepSeqIdPeriod;
70  
71    //TODO: depending on Store is not good but, realistically, all compactors currently do.
72    Compactor(final Configuration conf, final Store store) {
73      this.conf = conf;
74      this.store = store;
75      this.compactionKVMax =
76        this.conf.getInt(HConstants.COMPACTION_KV_MAX, HConstants.COMPACTION_KV_MAX_DEFAULT);
77      this.compactionCompression = (this.store.getFamily() == null) ?
78          Compression.Algorithm.NONE : this.store.getFamily().getCompactionCompression();
79      this.keepSeqIdPeriod = Math.max(this.conf.getInt(HConstants.KEEP_SEQID_PERIOD, 
80        HConstants.MIN_KEEP_SEQID_PERIOD), HConstants.MIN_KEEP_SEQID_PERIOD);
81    }
82  
83    public interface CellSink {
84      void append(Cell cell) throws IOException;
85    }
86  
87    public CompactionProgress getProgress() {
88      return this.progress;
89    }
90  
91    /** The sole reason this class exists is that java has no ref/out/pointer parameters. */
92    protected static class FileDetails {
93      /** Maximum key count after compaction (for blooms) */
94      public long maxKeyCount = 0;
95      /** Earliest put timestamp if major compaction */
96      public long earliestPutTs = HConstants.LATEST_TIMESTAMP;
97      /** The last key in the files we're compacting. */
98      public long maxSeqId = 0;
99      /** Latest memstore read point found in any of the involved files */
100     public long maxMVCCReadpoint = 0;
101     /** Max tags length**/
102     public int maxTagsLength = 0;
103     /** Min SeqId to keep during a major compaction **/
104     public long minSeqIdToKeep = 0;
105   }
106 
107   /**
108    * Extracts some details about the files to compact that are commonly needed by compactors.
109    * @param filesToCompact Files.
110    * @param allFiles Whether all files are included for compaction
111    * @return The result.
112    */
113   protected FileDetails getFileDetails(
114       Collection<StoreFile> filesToCompact, boolean allFiles) throws IOException {
115     FileDetails fd = new FileDetails();
116     long oldestHFileTimeStampToKeepMVCC = System.currentTimeMillis() - 
117       (1000L * 60 * 60 * 24 * this.keepSeqIdPeriod);  
118 
119     for (StoreFile file : filesToCompact) {
120       if(allFiles && (file.getModificationTimeStamp() < oldestHFileTimeStampToKeepMVCC)) {
121         // when isAllFiles is true, all files are compacted so we can calculate the smallest 
122         // MVCC value to keep
123         if(fd.minSeqIdToKeep < file.getMaxMemstoreTS()) {
124           fd.minSeqIdToKeep = file.getMaxMemstoreTS();
125         }
126       }
127       long seqNum = file.getMaxSequenceId();
128       fd.maxSeqId = Math.max(fd.maxSeqId, seqNum);
129       StoreFile.Reader r = file.getReader();
130       if (r == null) {
131         LOG.warn("Null reader for " + file.getPath());
132         continue;
133       }
134       // NOTE: use getEntries when compacting instead of getFilterEntries, otherwise under-sized
135       // blooms can cause progress to be miscalculated or if the user switches bloom
136       // type (e.g. from ROW to ROWCOL)
137       long keyCount = r.getEntries();
138       fd.maxKeyCount += keyCount;
139       // calculate the latest MVCC readpoint in any of the involved store files
140       Map<byte[], byte[]> fileInfo = r.loadFileInfo();
141       byte tmp[] = null;
142       // Get and set the real MVCCReadpoint for bulk loaded files, which is the
143       // SeqId number.
144       if (r.isBulkLoaded()) {
145         fd.maxMVCCReadpoint = Math.max(fd.maxMVCCReadpoint, r.getSequenceID());
146       }
147       else {
148         tmp = fileInfo.get(HFileWriterV2.MAX_MEMSTORE_TS_KEY);
149         if (tmp != null) {
150           fd.maxMVCCReadpoint = Math.max(fd.maxMVCCReadpoint, Bytes.toLong(tmp));
151         }
152       }
153       tmp = fileInfo.get(FileInfo.MAX_TAGS_LEN);
154       if (tmp != null) {
155         fd.maxTagsLength = Math.max(fd.maxTagsLength, Bytes.toInt(tmp));
156       }
157       // If required, calculate the earliest put timestamp of all involved storefiles.
158       // This is used to remove family delete marker during compaction.
159       long earliestPutTs = 0;
160       if (allFiles) {
161         tmp = fileInfo.get(StoreFile.EARLIEST_PUT_TS);
162         if (tmp == null) {
163           // There's a file with no information, must be an old one
164           // assume we have very old puts
165           fd.earliestPutTs = earliestPutTs = HConstants.OLDEST_TIMESTAMP;
166         } else {
167           earliestPutTs = Bytes.toLong(tmp);
168           fd.earliestPutTs = Math.min(fd.earliestPutTs, earliestPutTs);
169         }
170       }
171       if (LOG.isDebugEnabled()) {
172         LOG.debug("Compacting " + file +
173           ", keycount=" + keyCount +
174           ", bloomtype=" + r.getBloomFilterType().toString() +
175           ", size=" + TraditionalBinaryPrefix.long2String(r.length(), "", 1) +
176           ", encoding=" + r.getHFileReader().getDataBlockEncoding() +
177           ", seqNum=" + seqNum +
178           (allFiles ? ", earliestPutTs=" + earliestPutTs: ""));
179       }
180     }
181     return fd;
182   }
183 
184   /**
185    * Creates file scanners for compaction.
186    * @param filesToCompact Files.
187    * @return Scanners.
188    */
189   protected List<StoreFileScanner> createFileScanners(
190       final Collection<StoreFile> filesToCompact,
191       long smallestReadPoint,
192       boolean useDropBehind) throws IOException {
193     return StoreFileScanner.getScannersForStoreFiles(filesToCompact,
194         /* cache blocks = */ false,
195         /* use pread = */ false,
196         /* is compaction */ true,
197         /* use Drop Behind */ useDropBehind,
198       smallestReadPoint);
199   }
200 
201   protected long getSmallestReadPoint() {
202     return store.getSmallestReadPoint();
203   }
204 
205   /**
206    * Calls coprocessor, if any, to create compaction scanner - before normal scanner creation.
207    * @param request Compaction request.
208    * @param scanType Scan type.
209    * @param earliestPutTs Earliest put ts.
210    * @param scanners File scanners for compaction files.
211    * @return Scanner override by coprocessor; null if not overriding.
212    */
213   protected InternalScanner preCreateCoprocScanner(final CompactionRequest request,
214       ScanType scanType, long earliestPutTs,  List<StoreFileScanner> scanners) throws IOException {
215     return preCreateCoprocScanner(request, scanType, earliestPutTs, scanners, null);
216   }
217 
218   protected InternalScanner preCreateCoprocScanner(final CompactionRequest request,
219       final ScanType scanType, final long earliestPutTs, final List<StoreFileScanner> scanners,
220       User user) throws IOException {
221     if (store.getCoprocessorHost() == null) return null;
222     if (user == null) {
223       return store.getCoprocessorHost().preCompactScannerOpen(store, scanners, scanType,
224         earliestPutTs, request);
225     } else {
226       try {
227         return user.getUGI().doAs(new PrivilegedExceptionAction<InternalScanner>() {
228           @Override
229           public InternalScanner run() throws Exception {
230             return store.getCoprocessorHost().preCompactScannerOpen(store, scanners,
231               scanType, earliestPutTs, request);
232           }
233         });
234       } catch (InterruptedException ie) {
235         InterruptedIOException iioe = new InterruptedIOException();
236         iioe.initCause(ie);
237         throw iioe;
238       }
239     }
240   }
241 
242   /**
243    * Calls coprocessor, if any, to create scanners - after normal scanner creation.
244    * @param request Compaction request.
245    * @param scanType Scan type.
246    * @param scanner The default scanner created for compaction.
247    * @return Scanner scanner to use (usually the default); null if compaction should not proceed.
248    */
249    protected InternalScanner postCreateCoprocScanner(final CompactionRequest request,
250       final ScanType scanType, final InternalScanner scanner, User user) throws IOException {
251      if (store.getCoprocessorHost() == null) return scanner;
252      if (user == null) {
253        return store.getCoprocessorHost().preCompact(store, scanner, scanType, request);
254      } else {
255        try {
256          return user.getUGI().doAs(new PrivilegedExceptionAction<InternalScanner>() {
257            @Override
258            public InternalScanner run() throws Exception {
259              return store.getCoprocessorHost().preCompact(store, scanner, scanType, request);
260            }
261          });
262        } catch (InterruptedException ie) {
263          InterruptedIOException iioe = new InterruptedIOException();
264          iioe.initCause(ie);
265          throw iioe;
266        }
267      }
268   }
269 
270   /**
271    * Used to prevent compaction name conflict when multiple compactions running parallel on the
272    * same store.
273    */
274   private static final AtomicInteger NAME_COUNTER = new AtomicInteger(0);
275 
276   private String generateCompactionName() {
277     int counter;
278     for (;;) {
279       counter = NAME_COUNTER.get();
280       int next = counter == Integer.MAX_VALUE ? 0 : counter + 1;
281       if (NAME_COUNTER.compareAndSet(counter, next)) {
282         break;
283       }
284     }
285     return store.getRegionInfo().getRegionNameAsString() + "#"
286         + store.getFamily().getNameAsString() + "#" + counter;
287   }
288   /**
289    * Performs the compaction.
290    * @param scanner Where to read from.
291    * @param writer Where to write to.
292    * @param smallestReadPoint Smallest read point.
293    * @param cleanSeqId When true, remove seqId(used to be mvcc) value which is &lt;= smallestReadPoint
294    * @return Whether compaction ended; false if it was interrupted for some reason.
295    */
296   protected boolean performCompaction(InternalScanner scanner, CellSink writer,
297       long smallestReadPoint, boolean cleanSeqId,
298       CompactionThroughputController throughputController) throws IOException {
299     long bytesWritten = 0;
300     long bytesWrittenProgress = 0;
301     // Since scanner.next() can return 'false' but still be delivering data,
302     // we have to use a do/while loop.
303     List<Cell> cells = new ArrayList<Cell>();
304     long closeCheckInterval = HStore.getCloseCheckInterval();
305     long lastMillis = 0;
306     if (LOG.isDebugEnabled()) {
307       lastMillis = EnvironmentEdgeManager.currentTime();
308     }
309     String compactionName = generateCompactionName();
310     long now = 0;
311     boolean hasMore;
312     ScannerContext scannerContext =
313         ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
314 
315     throughputController.start(compactionName);
316     try {
317       do {
318         hasMore = scanner.next(cells, scannerContext);
319         if (LOG.isDebugEnabled()) {
320           now = EnvironmentEdgeManager.currentTime();
321         }
322         // output to writer:
323         Cell lastCleanCell = null;
324         long lastCleanCellSeqId = 0;
325         for (Cell c : cells) {
326           if (cleanSeqId && c.getSequenceId() <= smallestReadPoint) {
327             lastCleanCell = c;
328             lastCleanCellSeqId = c.getSequenceId();
329             CellUtil.setSequenceId(c, 0);
330           } else {
331             lastCleanCell = null;
332             lastCleanCellSeqId = 0;
333           }
334           writer.append(c);
335           int len = KeyValueUtil.length(c);
336           ++progress.currentCompactedKVs;
337           progress.totalCompactedSize += len;
338           if (LOG.isDebugEnabled()) {
339             bytesWrittenProgress += len;
340           }
341           throughputController.control(compactionName, len);
342           // check periodically to see if a system stop is requested
343           if (closeCheckInterval > 0) {
344             bytesWritten += len;
345             if (bytesWritten > closeCheckInterval) {
346               bytesWritten = 0;
347               if (!store.areWritesEnabled()) {
348                 progress.cancel();
349                 return false;
350               }
351             }
352           }
353         }
354         if (lastCleanCell != null) {
355           // HBASE-16931, set back sequence id to avoid affecting scan order unexpectedly
356           CellUtil.setSequenceId(lastCleanCell, lastCleanCellSeqId);
357         }
358         // Log the progress of long running compactions every minute if
359         // logging at DEBUG level
360         if (LOG.isDebugEnabled()) {
361           if ((now - lastMillis) >= 60 * 1000) {
362             LOG.debug("Compaction progress: "
363                 + compactionName
364                 + " "
365                 + progress
366                 + String.format(", rate=%.2f kB/sec", (bytesWrittenProgress / 1024.0)
367                     / ((now - lastMillis) / 1000.0)) + ", throughputController is "
368                 + throughputController);
369             lastMillis = now;
370             bytesWrittenProgress = 0;
371           }
372         }
373         cells.clear();
374       } while (hasMore);
375     } catch (InterruptedException e) {
376       progress.cancel();
377       throw new InterruptedIOException("Interrupted while control throughput of compacting "
378           + compactionName);
379     } finally {
380       throughputController.finish(compactionName);
381     }
382     progress.complete();
383     return true;
384   }
385 
386   /**
387    * @param store store
388    * @param scanners Store file scanners.
389    * @param scanType Scan type.
390    * @param smallestReadPoint Smallest MVCC read point.
391    * @param earliestPutTs Earliest put across all files.
392    * @return A compaction scanner.
393    */
394   protected InternalScanner createScanner(Store store, List<StoreFileScanner> scanners,
395       ScanType scanType, long smallestReadPoint, long earliestPutTs) throws IOException {
396     Scan scan = new Scan();
397     scan.setMaxVersions(store.getFamily().getMaxVersions());
398     return new StoreScanner(store, store.getScanInfo(), scan, scanners,
399         scanType, smallestReadPoint, earliestPutTs);
400   }
401 
402   /**
403    * @param store The store.
404    * @param scanners Store file scanners.
405    * @param smallestReadPoint Smallest MVCC read point.
406    * @param earliestPutTs Earliest put across all files.
407    * @param dropDeletesFromRow Drop deletes starting with this row, inclusive. Can be null.
408    * @param dropDeletesToRow Drop deletes ending with this row, exclusive. Can be null.
409    * @return A compaction scanner.
410    */
411   protected InternalScanner createScanner(Store store, List<StoreFileScanner> scanners,
412      long smallestReadPoint, long earliestPutTs, byte[] dropDeletesFromRow,
413      byte[] dropDeletesToRow) throws IOException {
414     Scan scan = new Scan();
415     scan.setMaxVersions(store.getFamily().getMaxVersions());
416     return new StoreScanner(store, store.getScanInfo(), scan, scanners, smallestReadPoint,
417         earliestPutTs, dropDeletesFromRow, dropDeletesToRow);
418   }
419 }