001/** 002 * 003 * Licensed to the Apache Software Foundation (ASF) under one 004 * or more contributor license agreements. See the NOTICE file 005 * distributed with this work for additional information 006 * regarding copyright ownership. The ASF licenses this file 007 * to you under the Apache License, Version 2.0 (the 008 * "License"); you may not use this file except in compliance 009 * with the License. You may obtain a copy of the License at 010 * 011 * http://www.apache.org/licenses/LICENSE-2.0 012 * 013 * Unless required by applicable law or agreed to in writing, software 014 * distributed under the License is distributed on an "AS IS" BASIS, 015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 016 * See the License for the specific language governing permissions and 017 * limitations under the License. 018 */ 019package org.apache.hadoop.hbase.regionserver; 020 021import java.io.IOException; 022import java.util.Collections; 023import java.util.Map; 024import java.util.Optional; 025import java.util.OptionalLong; 026import java.util.concurrent.atomic.AtomicBoolean; 027import java.util.concurrent.atomic.AtomicInteger; 028 029import org.apache.hadoop.conf.Configuration; 030import org.apache.hadoop.fs.FileSystem; 031import org.apache.hadoop.fs.Path; 032import org.apache.hadoop.hbase.Cell; 033import org.apache.hadoop.hbase.CellComparator; 034import org.apache.hadoop.hbase.HDFSBlocksDistribution; 035import org.apache.hadoop.hbase.io.TimeRange; 036import org.apache.hadoop.hbase.io.hfile.BlockType; 037import org.apache.hadoop.hbase.io.hfile.CacheConfig; 038import org.apache.hadoop.hbase.io.hfile.HFile; 039import org.apache.hadoop.hbase.util.BloomFilterFactory; 040import org.apache.hadoop.hbase.util.Bytes; 041import org.apache.yetus.audience.InterfaceAudience; 042import org.slf4j.Logger; 043import org.slf4j.LoggerFactory; 044import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 045 046/** 047 * A Store data file. Stores usually have one or more of these files. They 048 * are produced by flushing the memstore to disk. To 049 * create, instantiate a writer using {@link StoreFileWriter.Builder} 050 * and append data. Be sure to add any metadata before calling close on the 051 * Writer (Use the appendMetadata convenience methods). On close, a StoreFile 052 * is sitting in the Filesystem. To refer to it, create a StoreFile instance 053 * passing filesystem and path. To read, call {@link #initReader()} 054 * <p>StoreFiles may also reference store files in another Store. 055 * 056 * The reason for this weird pattern where you use a different instance for the 057 * writer and a reader is that we write once but read a lot more. 058 */ 059@InterfaceAudience.Private 060public class HStoreFile implements StoreFile { 061 062 private static final Logger LOG = LoggerFactory.getLogger(HStoreFile.class.getName()); 063 064 public static final String STORE_FILE_READER_NO_READAHEAD = "hbase.store.reader.no-readahead"; 065 066 private static final boolean DEFAULT_STORE_FILE_READER_NO_READAHEAD = false; 067 068 // Keys for fileinfo values in HFile 069 070 /** Max Sequence ID in FileInfo */ 071 public static final byte[] MAX_SEQ_ID_KEY = Bytes.toBytes("MAX_SEQ_ID_KEY"); 072 073 /** Major compaction flag in FileInfo */ 074 public static final byte[] MAJOR_COMPACTION_KEY = Bytes.toBytes("MAJOR_COMPACTION_KEY"); 075 076 /** Minor compaction flag in FileInfo */ 077 public static final byte[] EXCLUDE_FROM_MINOR_COMPACTION_KEY = 078 Bytes.toBytes("EXCLUDE_FROM_MINOR_COMPACTION"); 079 080 /** Bloom filter Type in FileInfo */ 081 public static final byte[] BLOOM_FILTER_TYPE_KEY = Bytes.toBytes("BLOOM_FILTER_TYPE"); 082 083 /** Delete Family Count in FileInfo */ 084 public static final byte[] DELETE_FAMILY_COUNT = Bytes.toBytes("DELETE_FAMILY_COUNT"); 085 086 /** Last Bloom filter key in FileInfo */ 087 public static final byte[] LAST_BLOOM_KEY = Bytes.toBytes("LAST_BLOOM_KEY"); 088 089 /** Key for Timerange information in metadata */ 090 public static final byte[] TIMERANGE_KEY = Bytes.toBytes("TIMERANGE"); 091 092 /** Key for timestamp of earliest-put in metadata */ 093 public static final byte[] EARLIEST_PUT_TS = Bytes.toBytes("EARLIEST_PUT_TS"); 094 095 /** Key for the number of mob cells in metadata */ 096 public static final byte[] MOB_CELLS_COUNT = Bytes.toBytes("MOB_CELLS_COUNT"); 097 098 /** Meta key set when store file is a result of a bulk load */ 099 public static final byte[] BULKLOAD_TASK_KEY = Bytes.toBytes("BULKLOAD_SOURCE_TASK"); 100 public static final byte[] BULKLOAD_TIME_KEY = Bytes.toBytes("BULKLOAD_TIMESTAMP"); 101 102 /** 103 * Key for skipping resetting sequence id in metadata. For bulk loaded hfiles, the scanner resets 104 * the cell seqId with the latest one, if this metadata is set as true, the reset is skipped. 105 */ 106 public static final byte[] SKIP_RESET_SEQ_ID = Bytes.toBytes("SKIP_RESET_SEQ_ID"); 107 108 private final StoreFileInfo fileInfo; 109 private final FileSystem fs; 110 111 // Block cache configuration and reference. 112 private final CacheConfig cacheConf; 113 114 // Counter that is incremented every time a scanner is created on the 115 // store file. It is decremented when the scan on the store file is 116 // done. 117 private final AtomicInteger refCount = new AtomicInteger(0); 118 119 private final boolean noReadahead; 120 121 private final boolean primaryReplica; 122 123 // Indicates if the file got compacted 124 private volatile boolean compactedAway = false; 125 126 // Keys for metadata stored in backing HFile. 127 // Set when we obtain a Reader. 128 private long sequenceid = -1; 129 130 // max of the MemstoreTS in the KV's in this store 131 // Set when we obtain a Reader. 132 private long maxMemstoreTS = -1; 133 134 // firstKey, lastkey and cellComparator will be set when openReader. 135 private Optional<Cell> firstKey; 136 137 private Optional<Cell> lastKey; 138 139 private CellComparator comparator; 140 141 public CacheConfig getCacheConf() { 142 return cacheConf; 143 } 144 145 @Override 146 public Optional<Cell> getFirstKey() { 147 return firstKey; 148 } 149 150 @Override 151 public Optional<Cell> getLastKey() { 152 return lastKey; 153 } 154 155 @Override 156 public CellComparator getComparator() { 157 return comparator; 158 } 159 160 @Override 161 public long getMaxMemStoreTS() { 162 return maxMemstoreTS; 163 } 164 165 // If true, this file was product of a major compaction. Its then set 166 // whenever you get a Reader. 167 private AtomicBoolean majorCompaction = null; 168 169 // If true, this file should not be included in minor compactions. 170 // It's set whenever you get a Reader. 171 private boolean excludeFromMinorCompaction = false; 172 173 /** 174 * Map of the metadata entries in the corresponding HFile. Populated when Reader is opened 175 * after which it is not modified again. 176 */ 177 private Map<byte[], byte[]> metadataMap; 178 179 // StoreFile.Reader 180 private volatile StoreFileReader reader; 181 182 /** 183 * Bloom filter type specified in column family configuration. Does not 184 * necessarily correspond to the Bloom filter type present in the HFile. 185 */ 186 private final BloomType cfBloomType; 187 188 /** 189 * Constructor, loads a reader and it's indices, etc. May allocate a substantial amount of ram 190 * depending on the underlying files (10-20MB?). 191 * @param fs The current file system to use. 192 * @param p The path of the file. 193 * @param conf The current configuration. 194 * @param cacheConf The cache configuration and block cache reference. 195 * @param cfBloomType The bloom type to use for this store file as specified by column family 196 * configuration. This may or may not be the same as the Bloom filter type actually 197 * present in the HFile, because column family configuration might change. If this is 198 * {@link BloomType#NONE}, the existing Bloom filter is ignored. 199 * @param primaryReplica true if this is a store file for primary replica, otherwise false. 200 * @throws IOException 201 */ 202 public HStoreFile(FileSystem fs, Path p, Configuration conf, CacheConfig cacheConf, 203 BloomType cfBloomType, boolean primaryReplica) throws IOException { 204 this(fs, new StoreFileInfo(conf, fs, p), conf, cacheConf, cfBloomType, primaryReplica); 205 } 206 207 /** 208 * Constructor, loads a reader and it's indices, etc. May allocate a substantial amount of ram 209 * depending on the underlying files (10-20MB?). 210 * @param fs fs The current file system to use. 211 * @param fileInfo The store file information. 212 * @param conf The current configuration. 213 * @param cacheConf The cache configuration and block cache reference. 214 * @param cfBloomType The bloom type to use for this store file as specified by column 215 * family configuration. This may or may not be the same as the Bloom filter type 216 * actually present in the HFile, because column family configuration might change. If 217 * this is {@link BloomType#NONE}, the existing Bloom filter is ignored. 218 * @param primaryReplica true if this is a store file for primary replica, otherwise false. 219 */ 220 public HStoreFile(FileSystem fs, StoreFileInfo fileInfo, Configuration conf, CacheConfig cacheConf, 221 BloomType cfBloomType, boolean primaryReplica) { 222 this.fs = fs; 223 this.fileInfo = fileInfo; 224 this.cacheConf = cacheConf; 225 this.noReadahead = 226 conf.getBoolean(STORE_FILE_READER_NO_READAHEAD, DEFAULT_STORE_FILE_READER_NO_READAHEAD); 227 if (BloomFilterFactory.isGeneralBloomEnabled(conf)) { 228 this.cfBloomType = cfBloomType; 229 } else { 230 LOG.info("Ignoring bloom filter check for file " + this.getPath() + ": " + "cfBloomType=" + 231 cfBloomType + " (disabled in config)"); 232 this.cfBloomType = BloomType.NONE; 233 } 234 this.primaryReplica = primaryReplica; 235 } 236 237 /** 238 * @return the StoreFile object associated to this StoreFile. null if the StoreFile is not a 239 * reference. 240 */ 241 public StoreFileInfo getFileInfo() { 242 return this.fileInfo; 243 } 244 245 @Override 246 public Path getPath() { 247 return this.fileInfo.getPath(); 248 } 249 250 @Override 251 public Path getQualifiedPath() { 252 return this.fileInfo.getPath().makeQualified(fs.getUri(), fs.getWorkingDirectory()); 253 } 254 255 @Override 256 public boolean isReference() { 257 return this.fileInfo.isReference(); 258 } 259 260 @Override 261 public boolean isHFile() { 262 return StoreFileInfo.isHFile(this.fileInfo.getPath()); 263 } 264 265 @Override 266 public boolean isMajorCompactionResult() { 267 if (this.majorCompaction == null) { 268 throw new NullPointerException("This has not been set yet"); 269 } 270 return this.majorCompaction.get(); 271 } 272 273 @Override 274 public boolean excludeFromMinorCompaction() { 275 return this.excludeFromMinorCompaction; 276 } 277 278 @Override 279 public long getMaxSequenceId() { 280 return this.sequenceid; 281 } 282 283 @Override 284 public long getModificationTimeStamp() throws IOException { 285 return getModificationTimestamp(); 286 } 287 288 @Override 289 public long getModificationTimestamp() throws IOException { 290 return fileInfo.getModificationTime(); 291 } 292 293 /** 294 * Only used by the Striped Compaction Policy 295 * @param key 296 * @return value associated with the metadata key 297 */ 298 public byte[] getMetadataValue(byte[] key) { 299 return metadataMap.get(key); 300 } 301 302 @Override 303 public boolean isBulkLoadResult() { 304 boolean bulkLoadedHFile = false; 305 String fileName = this.getPath().getName(); 306 int startPos = fileName.indexOf("SeqId_"); 307 if (startPos != -1) { 308 bulkLoadedHFile = true; 309 } 310 return bulkLoadedHFile || (metadataMap != null && metadataMap.containsKey(BULKLOAD_TIME_KEY)); 311 } 312 313 public boolean isCompactedAway() { 314 return compactedAway; 315 } 316 317 @VisibleForTesting 318 public int getRefCount() { 319 return refCount.get(); 320 } 321 322 /** 323 * @return true if the file is still used in reads 324 */ 325 public boolean isReferencedInReads() { 326 int rc = refCount.get(); 327 assert rc >= 0; // we should not go negative. 328 return rc > 0; 329 } 330 331 @Override 332 public OptionalLong getBulkLoadTimestamp() { 333 byte[] bulkLoadTimestamp = metadataMap.get(BULKLOAD_TIME_KEY); 334 return bulkLoadTimestamp == null ? OptionalLong.empty() 335 : OptionalLong.of(Bytes.toLong(bulkLoadTimestamp)); 336 } 337 338 /** 339 * @return the cached value of HDFS blocks distribution. The cached value is calculated when store 340 * file is opened. 341 */ 342 public HDFSBlocksDistribution getHDFSBlockDistribution() { 343 return this.fileInfo.getHDFSBlockDistribution(); 344 } 345 346 /** 347 * Opens reader on this store file. Called by Constructor. 348 * @throws IOException 349 * @see #closeStoreFile(boolean) 350 */ 351 private void open() throws IOException { 352 if (this.reader != null) { 353 throw new IllegalAccessError("Already open"); 354 } 355 356 // Open the StoreFile.Reader 357 this.reader = fileInfo.open(this.fs, this.cacheConf, false, noReadahead ? 0L : -1L, 358 primaryReplica, refCount, true); 359 360 // Load up indices and fileinfo. This also loads Bloom filter type. 361 metadataMap = Collections.unmodifiableMap(this.reader.loadFileInfo()); 362 363 // Read in our metadata. 364 byte [] b = metadataMap.get(MAX_SEQ_ID_KEY); 365 if (b != null) { 366 // By convention, if halfhfile, top half has a sequence number > bottom 367 // half. Thats why we add one in below. Its done for case the two halves 368 // are ever merged back together --rare. Without it, on open of store, 369 // since store files are distinguished by sequence id, the one half would 370 // subsume the other. 371 this.sequenceid = Bytes.toLong(b); 372 if (fileInfo.isTopReference()) { 373 this.sequenceid += 1; 374 } 375 } 376 377 if (isBulkLoadResult()){ 378 // generate the sequenceId from the fileName 379 // fileName is of the form <randomName>_SeqId_<id-when-loaded>_ 380 String fileName = this.getPath().getName(); 381 // Use lastIndexOf() to get the last, most recent bulk load seqId. 382 int startPos = fileName.lastIndexOf("SeqId_"); 383 if (startPos != -1) { 384 this.sequenceid = Long.parseLong(fileName.substring(startPos + 6, 385 fileName.indexOf('_', startPos + 6))); 386 // Handle reference files as done above. 387 if (fileInfo.isTopReference()) { 388 this.sequenceid += 1; 389 } 390 } 391 // SKIP_RESET_SEQ_ID only works in bulk loaded file. 392 // In mob compaction, the hfile where the cells contain the path of a new mob file is bulk 393 // loaded to hbase, these cells have the same seqIds with the old ones. We do not want 394 // to reset new seqIds for them since this might make a mess of the visibility of cells that 395 // have the same row key but different seqIds. 396 boolean skipResetSeqId = isSkipResetSeqId(metadataMap.get(SKIP_RESET_SEQ_ID)); 397 if (skipResetSeqId) { 398 // increase the seqId when it is a bulk loaded file from mob compaction. 399 this.sequenceid += 1; 400 } 401 this.reader.setSkipResetSeqId(skipResetSeqId); 402 this.reader.setBulkLoaded(true); 403 } 404 this.reader.setSequenceID(this.sequenceid); 405 406 b = metadataMap.get(HFile.Writer.MAX_MEMSTORE_TS_KEY); 407 if (b != null) { 408 this.maxMemstoreTS = Bytes.toLong(b); 409 } 410 411 b = metadataMap.get(MAJOR_COMPACTION_KEY); 412 if (b != null) { 413 boolean mc = Bytes.toBoolean(b); 414 if (this.majorCompaction == null) { 415 this.majorCompaction = new AtomicBoolean(mc); 416 } else { 417 this.majorCompaction.set(mc); 418 } 419 } else { 420 // Presume it is not major compacted if it doesn't explicity say so 421 // HFileOutputFormat explicitly sets the major compacted key. 422 this.majorCompaction = new AtomicBoolean(false); 423 } 424 425 b = metadataMap.get(EXCLUDE_FROM_MINOR_COMPACTION_KEY); 426 this.excludeFromMinorCompaction = (b != null && Bytes.toBoolean(b)); 427 428 BloomType hfileBloomType = reader.getBloomFilterType(); 429 if (cfBloomType != BloomType.NONE) { 430 reader.loadBloomfilter(BlockType.GENERAL_BLOOM_META); 431 if (hfileBloomType != cfBloomType) { 432 LOG.info("HFile Bloom filter type for " 433 + reader.getHFileReader().getName() + ": " + hfileBloomType 434 + ", but " + cfBloomType + " specified in column family " 435 + "configuration"); 436 } 437 } else if (hfileBloomType != BloomType.NONE) { 438 LOG.info("Bloom filter turned off by CF config for " 439 + reader.getHFileReader().getName()); 440 } 441 442 // load delete family bloom filter 443 reader.loadBloomfilter(BlockType.DELETE_FAMILY_BLOOM_META); 444 445 try { 446 byte[] data = metadataMap.get(TIMERANGE_KEY); 447 this.reader.timeRange = data == null ? null : TimeRangeTracker.parseFrom(data).toTimeRange(); 448 } catch (IllegalArgumentException e) { 449 LOG.error("Error reading timestamp range data from meta -- " + 450 "proceeding without", e); 451 this.reader.timeRange = null; 452 } 453 // initialize so we can reuse them after reader closed. 454 firstKey = reader.getFirstKey(); 455 lastKey = reader.getLastKey(); 456 comparator = reader.getComparator(); 457 } 458 459 /** 460 * Initialize the reader used for pread. 461 */ 462 public void initReader() throws IOException { 463 if (reader == null) { 464 try { 465 open(); 466 } catch (Exception e) { 467 try { 468 boolean evictOnClose = cacheConf != null ? cacheConf.shouldEvictOnClose() : true; 469 this.closeStoreFile(evictOnClose); 470 } catch (IOException ee) { 471 LOG.warn("failed to close reader", ee); 472 } 473 throw e; 474 } 475 } 476 } 477 478 private StoreFileReader createStreamReader(boolean canUseDropBehind) throws IOException { 479 initReader(); 480 StoreFileReader reader = fileInfo.open(this.fs, this.cacheConf, canUseDropBehind, -1L, 481 primaryReplica, refCount, false); 482 reader.copyFields(this.reader); 483 return reader; 484 } 485 486 /** 487 * Get a scanner which uses pread. 488 * <p> 489 * Must be called after initReader. 490 */ 491 public StoreFileScanner getPreadScanner(boolean cacheBlocks, long readPt, long scannerOrder, 492 boolean canOptimizeForNonNullColumn) { 493 return getReader().getStoreFileScanner(cacheBlocks, true, false, readPt, scannerOrder, 494 canOptimizeForNonNullColumn); 495 } 496 497 /** 498 * Get a scanner which uses streaming read. 499 * <p> 500 * Must be called after initReader. 501 */ 502 public StoreFileScanner getStreamScanner(boolean canUseDropBehind, boolean cacheBlocks, 503 boolean isCompaction, long readPt, long scannerOrder, boolean canOptimizeForNonNullColumn) 504 throws IOException { 505 return createStreamReader(canUseDropBehind).getStoreFileScanner(cacheBlocks, false, 506 isCompaction, readPt, scannerOrder, canOptimizeForNonNullColumn); 507 } 508 509 /** 510 * @return Current reader. Must call initReader first else returns null. 511 * @see #initReader() 512 */ 513 public StoreFileReader getReader() { 514 return this.reader; 515 } 516 517 /** 518 * @param evictOnClose whether to evict blocks belonging to this file 519 * @throws IOException 520 */ 521 public synchronized void closeStoreFile(boolean evictOnClose) throws IOException { 522 if (this.reader != null) { 523 this.reader.close(evictOnClose); 524 this.reader = null; 525 } 526 } 527 528 /** 529 * Delete this file 530 * @throws IOException 531 */ 532 public void deleteStoreFile() throws IOException { 533 boolean evictOnClose = cacheConf != null ? cacheConf.shouldEvictOnClose() : true; 534 closeStoreFile(evictOnClose); 535 this.fs.delete(getPath(), true); 536 } 537 538 public void markCompactedAway() { 539 this.compactedAway = true; 540 } 541 542 @Override 543 public String toString() { 544 return this.fileInfo.toString(); 545 } 546 547 @Override 548 public String toStringDetailed() { 549 StringBuilder sb = new StringBuilder(); 550 sb.append(this.getPath().toString()); 551 sb.append(", isReference=").append(isReference()); 552 sb.append(", isBulkLoadResult=").append(isBulkLoadResult()); 553 if (isBulkLoadResult()) { 554 sb.append(", bulkLoadTS="); 555 OptionalLong bulkLoadTS = getBulkLoadTimestamp(); 556 if (bulkLoadTS.isPresent()) { 557 sb.append(bulkLoadTS.getAsLong()); 558 } else { 559 sb.append("NotPresent"); 560 } 561 } else { 562 sb.append(", seqid=").append(getMaxSequenceId()); 563 } 564 sb.append(", majorCompaction=").append(isMajorCompactionResult()); 565 566 return sb.toString(); 567 } 568 569 /** 570 * Gets whether to skip resetting the sequence id for cells. 571 * @param skipResetSeqId The byte array of boolean. 572 * @return Whether to skip resetting the sequence id. 573 */ 574 private boolean isSkipResetSeqId(byte[] skipResetSeqId) { 575 if (skipResetSeqId != null && skipResetSeqId.length == 1) { 576 return Bytes.toBoolean(skipResetSeqId); 577 } 578 return false; 579 } 580 581 @Override 582 public OptionalLong getMinimumTimestamp() { 583 TimeRange tr = getReader().timeRange; 584 return tr != null ? OptionalLong.of(tr.getMin()) : OptionalLong.empty(); 585 } 586 587 @Override 588 public OptionalLong getMaximumTimestamp() { 589 TimeRange tr = getReader().timeRange; 590 return tr != null ? OptionalLong.of(tr.getMax()) : OptionalLong.empty(); 591 } 592}