001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.regionserver; 019 020import com.google.errorprone.annotations.RestrictedApi; 021import java.io.IOException; 022import java.io.InterruptedIOException; 023import java.net.InetSocketAddress; 024import java.util.ArrayList; 025import java.util.Collection; 026import java.util.Collections; 027import java.util.HashMap; 028import java.util.HashSet; 029import java.util.Iterator; 030import java.util.List; 031import java.util.Map; 032import java.util.Map.Entry; 033import java.util.NavigableSet; 034import java.util.Optional; 035import java.util.OptionalDouble; 036import java.util.OptionalInt; 037import java.util.OptionalLong; 038import java.util.Set; 039import java.util.concurrent.Callable; 040import java.util.concurrent.CompletionService; 041import java.util.concurrent.ConcurrentHashMap; 042import java.util.concurrent.ExecutionException; 043import java.util.concurrent.ExecutorCompletionService; 044import java.util.concurrent.Future; 045import java.util.concurrent.ThreadPoolExecutor; 046import java.util.concurrent.atomic.AtomicBoolean; 047import java.util.concurrent.atomic.AtomicInteger; 048import java.util.concurrent.atomic.AtomicLong; 049import java.util.concurrent.atomic.LongAdder; 050import java.util.concurrent.locks.ReentrantLock; 051import java.util.function.Consumer; 052import java.util.function.Supplier; 053import java.util.function.ToLongFunction; 054import java.util.stream.Collectors; 055import java.util.stream.LongStream; 056import org.apache.hadoop.conf.Configuration; 057import org.apache.hadoop.fs.FileSystem; 058import org.apache.hadoop.fs.Path; 059import org.apache.hadoop.fs.permission.FsAction; 060import org.apache.hadoop.hbase.Cell; 061import org.apache.hadoop.hbase.CellComparator; 062import org.apache.hadoop.hbase.CellUtil; 063import org.apache.hadoop.hbase.ExtendedCell; 064import org.apache.hadoop.hbase.HConstants; 065import org.apache.hadoop.hbase.InnerStoreCellComparator; 066import org.apache.hadoop.hbase.MemoryCompactionPolicy; 067import org.apache.hadoop.hbase.MetaCellComparator; 068import org.apache.hadoop.hbase.TableName; 069import org.apache.hadoop.hbase.backup.FailedArchiveException; 070import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; 071import org.apache.hadoop.hbase.client.RegionInfo; 072import org.apache.hadoop.hbase.client.Scan; 073import org.apache.hadoop.hbase.conf.ConfigKey; 074import org.apache.hadoop.hbase.conf.ConfigurationManager; 075import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver; 076import org.apache.hadoop.hbase.coprocessor.ReadOnlyConfiguration; 077import org.apache.hadoop.hbase.io.HeapSize; 078import org.apache.hadoop.hbase.io.hfile.CacheConfig; 079import org.apache.hadoop.hbase.io.hfile.HFile; 080import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder; 081import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoderImpl; 082import org.apache.hadoop.hbase.io.hfile.HFileScanner; 083import org.apache.hadoop.hbase.io.hfile.InvalidHFileException; 084import org.apache.hadoop.hbase.monitoring.MonitoredTask; 085import org.apache.hadoop.hbase.quotas.RegionSizeStore; 086import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext; 087import org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker; 088import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress; 089import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl; 090import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours; 091import org.apache.hadoop.hbase.regionserver.querymatcher.ScanQueryMatcher; 092import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker; 093import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory; 094import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController; 095import org.apache.hadoop.hbase.regionserver.wal.WALUtil; 096import org.apache.hadoop.hbase.security.EncryptionUtil; 097import org.apache.hadoop.hbase.security.User; 098import org.apache.hadoop.hbase.util.Bytes; 099import org.apache.hadoop.hbase.util.ClassSize; 100import org.apache.hadoop.hbase.util.CommonFSUtils; 101import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 102import org.apache.hadoop.hbase.util.Pair; 103import org.apache.hadoop.hbase.util.ReflectionUtils; 104import org.apache.hadoop.util.StringUtils; 105import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix; 106import org.apache.yetus.audience.InterfaceAudience; 107import org.slf4j.Logger; 108import org.slf4j.LoggerFactory; 109 110import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; 111import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableCollection; 112import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableList; 113import org.apache.hbase.thirdparty.com.google.common.collect.Lists; 114import org.apache.hbase.thirdparty.com.google.common.collect.Maps; 115import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils; 116import org.apache.hbase.thirdparty.org.apache.commons.collections4.IterableUtils; 117 118import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 119import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.CompactionDescriptor; 120 121/** 122 * A Store holds a column family in a Region. Its a memstore and a set of zero or more StoreFiles, 123 * which stretch backwards over time. 124 * <p> 125 * There's no reason to consider append-logging at this level; all logging and locking is handled at 126 * the HRegion level. Store just provides services to manage sets of StoreFiles. One of the most 127 * important of those services is compaction services where files are aggregated once they pass a 128 * configurable threshold. 129 * <p> 130 * Locking and transactions are handled at a higher level. This API should not be called directly 131 * but by an HRegion manager. 132 */ 133@InterfaceAudience.Private 134public class HStore 135 implements Store, HeapSize, StoreConfigInformation, PropagatingConfigurationObserver { 136 public static final String MEMSTORE_CLASS_NAME = 137 ConfigKey.CLASS("hbase.regionserver.memstore.class", MemStore.class); 138 public static final String COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY = 139 ConfigKey.INT("hbase.server.compactchecker.interval.multiplier"); 140 public static final String BLOCKING_STOREFILES_KEY = 141 ConfigKey.INT("hbase.hstore.blockingStoreFiles"); 142 public static final String BLOCK_STORAGE_POLICY_KEY = "hbase.hstore.block.storage.policy"; 143 // "NONE" is not a valid storage policy and means we defer the policy to HDFS 144 public static final String DEFAULT_BLOCK_STORAGE_POLICY = "NONE"; 145 public static final int DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER = 1000; 146 public static final int DEFAULT_BLOCKING_STOREFILE_COUNT = 16; 147 148 // HBASE-24428 : Update compaction priority for recently split daughter regions 149 // so as to prioritize their compaction. 150 // Any compaction candidate with higher priority than compaction of newly split daugher regions 151 // should have priority value < (Integer.MIN_VALUE + 1000) 152 private static final int SPLIT_REGION_COMPACTION_PRIORITY = Integer.MIN_VALUE + 1000; 153 154 private static final Logger LOG = LoggerFactory.getLogger(HStore.class); 155 156 protected final MemStore memstore; 157 // This stores directory in the filesystem. 158 private final HRegion region; 159 protected Configuration conf; 160 private long lastCompactSize = 0; 161 volatile boolean forceMajor = false; 162 private AtomicLong storeSize = new AtomicLong(); 163 private AtomicLong totalUncompressedBytes = new AtomicLong(); 164 private LongAdder memstoreOnlyRowReadsCount = new LongAdder(); 165 // rows that has cells from both memstore and files (or only files) 166 private LongAdder mixedRowReadsCount = new LongAdder(); 167 168 /** 169 * Lock specific to archiving compacted store files. This avoids races around the combination of 170 * retrieving the list of compacted files and moving them to the archive directory. Since this is 171 * usually a background process (other than on close), we don't want to handle this with the store 172 * write lock, which would block readers and degrade performance. Locked by: - 173 * CompactedHFilesDispatchHandler via closeAndArchiveCompactedFiles() - close() 174 */ 175 final ReentrantLock archiveLock = new ReentrantLock(); 176 177 private final boolean verifyBulkLoads; 178 179 /** 180 * Use this counter to track concurrent puts. If TRACE-log is enabled, if we are over the 181 * threshold set by hbase.region.store.parallel.put.print.threshold (Default is 50) we will log a 182 * message that identifies the Store experience this high-level of concurrency. 183 */ 184 private final AtomicInteger currentParallelPutCount = new AtomicInteger(0); 185 private final int parallelPutCountPrintThreshold; 186 187 private ScanInfo scanInfo; 188 189 // All access must be synchronized. 190 // TODO: ideally, this should be part of storeFileManager, as we keep passing this to it. 191 private final List<HStoreFile> filesCompacting = Lists.newArrayList(); 192 193 // All access must be synchronized. 194 private final Set<ChangedReadersObserver> changedReaderObservers = 195 Collections.newSetFromMap(new ConcurrentHashMap<ChangedReadersObserver, Boolean>()); 196 197 private HFileDataBlockEncoder dataBlockEncoder; 198 199 final StoreEngine<?, ?, ?, ?> storeEngine; 200 201 private static final AtomicBoolean offPeakCompactionTracker = new AtomicBoolean(); 202 private volatile OffPeakHours offPeakHours; 203 204 private static final int DEFAULT_FLUSH_RETRIES_NUMBER = 10; 205 private int flushRetriesNumber; 206 private int pauseTime; 207 208 private long blockingFileCount; 209 private int compactionCheckMultiplier; 210 211 private AtomicLong flushedCellsCount = new AtomicLong(); 212 private AtomicLong compactedCellsCount = new AtomicLong(); 213 private AtomicLong majorCompactedCellsCount = new AtomicLong(); 214 private AtomicLong flushedCellsSize = new AtomicLong(); 215 private AtomicLong flushedOutputFileSize = new AtomicLong(); 216 private AtomicLong compactedCellsSize = new AtomicLong(); 217 private AtomicLong majorCompactedCellsSize = new AtomicLong(); 218 219 private final StoreContext storeContext; 220 221 // Used to track the store files which are currently being written. For compaction, if we want to 222 // compact store file [a, b, c] to [d], then here we will record 'd'. And we will also use it to 223 // track the store files being written when flushing. 224 // Notice that the creation is in the background compaction or flush thread and we will get the 225 // files in other thread, so it needs to be thread safe. 226 private static final class StoreFileWriterCreationTracker implements Consumer<Path> { 227 228 private final Set<Path> files = Collections.newSetFromMap(new ConcurrentHashMap<>()); 229 230 @Override 231 public void accept(Path t) { 232 files.add(t); 233 } 234 235 public Set<Path> get() { 236 return Collections.unmodifiableSet(files); 237 } 238 } 239 240 // We may have multiple compaction running at the same time, and flush can also happen at the same 241 // time, so here we need to use a collection, and the collection needs to be thread safe. 242 // The implementation of StoreFileWriterCreationTracker is very simple and we will not likely to 243 // implement hashCode or equals for it, so here we just use ConcurrentHashMap. Changed to 244 // IdentityHashMap if later we want to implement hashCode or equals. 245 private final Set<StoreFileWriterCreationTracker> storeFileWriterCreationTrackers = 246 Collections.newSetFromMap(new ConcurrentHashMap<>()); 247 248 // For the SFT implementation which we will write tmp store file first, we do not need to clean up 249 // the broken store files under the data directory, which means we do not need to track the store 250 // file writer creation. So here we abstract a factory to return different trackers for different 251 // SFT implementations. 252 private final Supplier<StoreFileWriterCreationTracker> storeFileWriterCreationTrackerFactory; 253 254 private final boolean warmup; 255 256 /** 257 * Constructor 258 * @param family HColumnDescriptor for this column 259 * @param confParam configuration object failed. Can be null. 260 */ 261 protected HStore(final HRegion region, final ColumnFamilyDescriptor family, 262 final Configuration confParam, boolean warmup) throws IOException { 263 this.conf = StoreUtils.createStoreConfiguration(confParam, region.getTableDescriptor(), family); 264 265 this.region = region; 266 this.storeContext = initializeStoreContext(family); 267 268 // Assemble the store's home directory and Ensure it exists. 269 region.getRegionFileSystem().createStoreDir(family.getNameAsString()); 270 271 // set block storage policy for store directory 272 String policyName = family.getStoragePolicy(); 273 if (null == policyName) { 274 policyName = this.conf.get(BLOCK_STORAGE_POLICY_KEY, DEFAULT_BLOCK_STORAGE_POLICY); 275 } 276 region.getRegionFileSystem().setStoragePolicy(family.getNameAsString(), policyName.trim()); 277 278 this.dataBlockEncoder = new HFileDataBlockEncoderImpl(family.getDataBlockEncoding()); 279 280 // used by ScanQueryMatcher 281 long timeToPurgeDeletes = Math.max(conf.getLong("hbase.hstore.time.to.purge.deletes", 0), 0); 282 LOG.trace("Time to purge deletes set to {}ms in {}", timeToPurgeDeletes, this); 283 // Get TTL 284 long ttl = determineTTLFromFamily(family); 285 // Why not just pass a HColumnDescriptor in here altogether? Even if have 286 // to clone it? 287 scanInfo = 288 new ScanInfo(conf, family, ttl, timeToPurgeDeletes, this.storeContext.getComparator()); 289 this.memstore = getMemstore(); 290 291 this.offPeakHours = OffPeakHours.getInstance(conf); 292 293 this.verifyBulkLoads = conf.getBoolean("hbase.hstore.bulkload.verify", false); 294 295 this.blockingFileCount = conf.getInt(BLOCKING_STOREFILES_KEY, DEFAULT_BLOCKING_STOREFILE_COUNT); 296 this.compactionCheckMultiplier = conf.getInt(COMPACTCHECKER_INTERVAL_MULTIPLIER_KEY, 297 DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER); 298 if (this.compactionCheckMultiplier <= 0) { 299 LOG.error("Compaction check period multiplier must be positive, setting default: {}", 300 DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER); 301 this.compactionCheckMultiplier = DEFAULT_COMPACTCHECKER_INTERVAL_MULTIPLIER; 302 } 303 304 this.warmup = warmup; 305 this.storeEngine = createStoreEngine(this, this.conf, region.getCellComparator()); 306 storeEngine.initialize(warmup); 307 // if require writing to tmp dir first, then we just return null, which indicate that we do not 308 // need to track the creation of store file writer, otherwise we return a new 309 // StoreFileWriterCreationTracker. 310 this.storeFileWriterCreationTrackerFactory = storeEngine.requireWritingToTmpDirFirst() 311 ? () -> null 312 : () -> new StoreFileWriterCreationTracker(); 313 refreshStoreSizeAndTotalBytes(); 314 315 flushRetriesNumber = 316 conf.getInt("hbase.hstore.flush.retries.number", DEFAULT_FLUSH_RETRIES_NUMBER); 317 pauseTime = conf.getInt(HConstants.HBASE_SERVER_PAUSE, HConstants.DEFAULT_HBASE_SERVER_PAUSE); 318 if (flushRetriesNumber <= 0) { 319 throw new IllegalArgumentException( 320 "hbase.hstore.flush.retries.number must be > 0, not " + flushRetriesNumber); 321 } 322 323 int confPrintThreshold = 324 this.conf.getInt("hbase.region.store.parallel.put.print.threshold", 50); 325 if (confPrintThreshold < 10) { 326 confPrintThreshold = 10; 327 } 328 this.parallelPutCountPrintThreshold = confPrintThreshold; 329 330 LOG.info( 331 "Store={}, memstore type={}, storagePolicy={}, verifyBulkLoads={}, " 332 + "parallelPutCountPrintThreshold={}, encoding={}, compression={}", 333 this, memstore.getClass().getSimpleName(), policyName, verifyBulkLoads, 334 parallelPutCountPrintThreshold, family.getDataBlockEncoding(), family.getCompressionType()); 335 } 336 337 private StoreContext initializeStoreContext(ColumnFamilyDescriptor family) throws IOException { 338 return new StoreContext.Builder().withBlockSize(family.getBlocksize()) 339 .withEncryptionContext(EncryptionUtil.createEncryptionContext(conf, family)) 340 .withBloomType(family.getBloomFilterType()).withCacheConfig(createCacheConf(family)) 341 .withCellComparator(region.getTableDescriptor().isMetaTable() || conf 342 .getBoolean(HRegion.USE_META_CELL_COMPARATOR, HRegion.DEFAULT_USE_META_CELL_COMPARATOR) 343 ? MetaCellComparator.META_COMPARATOR 344 : InnerStoreCellComparator.INNER_STORE_COMPARATOR) 345 .withColumnFamilyDescriptor(family).withCompactedFilesSupplier(this::getCompactedFiles) 346 .withRegionFileSystem(region.getRegionFileSystem()) 347 .withFavoredNodesSupplier(this::getFavoredNodes) 348 .withFamilyStoreDirectoryPath( 349 region.getRegionFileSystem().getStoreDir(family.getNameAsString())) 350 .withRegionCoprocessorHost(region.getCoprocessorHost()).build(); 351 } 352 353 private InetSocketAddress[] getFavoredNodes() { 354 InetSocketAddress[] favoredNodes = null; 355 if (region.getRegionServerServices() != null) { 356 favoredNodes = region.getRegionServerServices() 357 .getFavoredNodesForRegion(region.getRegionInfo().getEncodedName()); 358 } 359 return favoredNodes; 360 } 361 362 /** Returns MemStore Instance to use in this store. */ 363 private MemStore getMemstore() { 364 MemStore ms = null; 365 // Check if in-memory-compaction configured. Note MemoryCompactionPolicy is an enum! 366 MemoryCompactionPolicy inMemoryCompaction = null; 367 if (this.getTableName().isSystemTable()) { 368 inMemoryCompaction = MemoryCompactionPolicy 369 .valueOf(conf.get("hbase.systemtables.compacting.memstore.type", "NONE").toUpperCase()); 370 } else { 371 inMemoryCompaction = getColumnFamilyDescriptor().getInMemoryCompaction(); 372 } 373 if (inMemoryCompaction == null) { 374 inMemoryCompaction = 375 MemoryCompactionPolicy.valueOf(conf.get(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY, 376 CompactingMemStore.COMPACTING_MEMSTORE_TYPE_DEFAULT).toUpperCase()); 377 } 378 379 switch (inMemoryCompaction) { 380 case NONE: 381 Class<? extends MemStore> memStoreClass = 382 conf.getClass(MEMSTORE_CLASS_NAME, DefaultMemStore.class, MemStore.class); 383 ms = ReflectionUtils.newInstance(memStoreClass, 384 new Object[] { conf, getComparator(), this.getHRegion().getRegionServicesForStores() }); 385 break; 386 default: 387 Class<? extends CompactingMemStore> compactingMemStoreClass = 388 conf.getClass(MEMSTORE_CLASS_NAME, CompactingMemStore.class, CompactingMemStore.class); 389 ms = 390 ReflectionUtils.newInstance(compactingMemStoreClass, new Object[] { conf, getComparator(), 391 this, this.getHRegion().getRegionServicesForStores(), inMemoryCompaction }); 392 } 393 return ms; 394 } 395 396 /** 397 * Creates the cache config. 398 * @param family The current column family. 399 */ 400 protected CacheConfig createCacheConf(final ColumnFamilyDescriptor family) { 401 CacheConfig cacheConf = new CacheConfig(conf, family, region.getBlockCache(), 402 region.getRegionServicesForStores().getByteBuffAllocator()); 403 LOG.info("Created cacheConfig: {}, for column family {} of region {} ", cacheConf, 404 family.getNameAsString(), region.getRegionInfo().getEncodedName()); 405 return cacheConf; 406 } 407 408 /** 409 * Creates the store engine configured for the given Store. 410 * @param store The store. An unfortunate dependency needed due to it being passed to 411 * coprocessors via the compactor. 412 * @param conf Store configuration. 413 * @param kvComparator KVComparator for storeFileManager. 414 * @return StoreEngine to use. 415 */ 416 protected StoreEngine<?, ?, ?, ?> createStoreEngine(HStore store, Configuration conf, 417 CellComparator kvComparator) throws IOException { 418 return StoreEngine.create(store, conf, kvComparator); 419 } 420 421 /** Returns TTL in seconds of the specified family */ 422 public static long determineTTLFromFamily(final ColumnFamilyDescriptor family) { 423 // HCD.getTimeToLive returns ttl in seconds. Convert to milliseconds. 424 long ttl = family.getTimeToLive(); 425 if (ttl == HConstants.FOREVER) { 426 // Default is unlimited ttl. 427 ttl = Long.MAX_VALUE; 428 } else if (ttl == -1) { 429 ttl = Long.MAX_VALUE; 430 } else { 431 // Second -> ms adjust for user data 432 ttl *= 1000; 433 } 434 return ttl; 435 } 436 437 public StoreContext getStoreContext() { 438 return storeContext; 439 } 440 441 @Override 442 public String getColumnFamilyName() { 443 return this.storeContext.getFamily().getNameAsString(); 444 } 445 446 @Override 447 public TableName getTableName() { 448 return this.getRegionInfo().getTable(); 449 } 450 451 @Override 452 public FileSystem getFileSystem() { 453 return storeContext.getRegionFileSystem().getFileSystem(); 454 } 455 456 public HRegionFileSystem getRegionFileSystem() { 457 return storeContext.getRegionFileSystem(); 458 } 459 460 /* Implementation of StoreConfigInformation */ 461 @Override 462 public long getStoreFileTtl() { 463 // TTL only applies if there's no MIN_VERSIONs setting on the column. 464 return (this.scanInfo.getMinVersions() == 0) ? this.scanInfo.getTtl() : Long.MAX_VALUE; 465 } 466 467 @Override 468 public long getMemStoreFlushSize() { 469 // TODO: Why is this in here? The flushsize of the region rather than the store? St.Ack 470 return this.region.memstoreFlushSize; 471 } 472 473 @Override 474 public MemStoreSize getFlushableSize() { 475 return this.memstore.getFlushableSize(); 476 } 477 478 @Override 479 public MemStoreSize getSnapshotSize() { 480 return this.memstore.getSnapshotSize(); 481 } 482 483 @Override 484 public long getCompactionCheckMultiplier() { 485 return this.compactionCheckMultiplier; 486 } 487 488 @Override 489 public long getBlockingFileCount() { 490 return blockingFileCount; 491 } 492 /* End implementation of StoreConfigInformation */ 493 494 @Override 495 public ColumnFamilyDescriptor getColumnFamilyDescriptor() { 496 return this.storeContext.getFamily(); 497 } 498 499 @Override 500 public OptionalLong getMaxSequenceId() { 501 return StoreUtils.getMaxSequenceIdInList(this.getStorefiles()); 502 } 503 504 @Override 505 public OptionalLong getMaxMemStoreTS() { 506 return StoreUtils.getMaxMemStoreTSInList(this.getStorefiles()); 507 } 508 509 /** Returns the data block encoder */ 510 public HFileDataBlockEncoder getDataBlockEncoder() { 511 return dataBlockEncoder; 512 } 513 514 /** 515 * Should be used only in tests. 516 * @param blockEncoder the block delta encoder to use 517 */ 518 void setDataBlockEncoderInTest(HFileDataBlockEncoder blockEncoder) { 519 this.dataBlockEncoder = blockEncoder; 520 } 521 522 private void postRefreshStoreFiles() throws IOException { 523 // Advance the memstore read point to be at least the new store files seqIds so that 524 // readers might pick it up. This assumes that the store is not getting any writes (otherwise 525 // in-flight transactions might be made visible) 526 getMaxSequenceId().ifPresent(region.getMVCC()::advanceTo); 527 refreshStoreSizeAndTotalBytes(); 528 } 529 530 @Override 531 public void refreshStoreFiles() throws IOException { 532 storeEngine.refreshStoreFiles(); 533 postRefreshStoreFiles(); 534 } 535 536 /** 537 * Replaces the store files that the store has with the given files. Mainly used by secondary 538 * region replicas to keep up to date with the primary region files. 539 */ 540 public void refreshStoreFiles(Collection<String> newFiles) throws IOException { 541 storeEngine.refreshStoreFiles(newFiles); 542 postRefreshStoreFiles(); 543 } 544 545 /** 546 * This message intends to inform the MemStore that next coming updates are going to be part of 547 * the replaying edits from WAL 548 */ 549 public void startReplayingFromWAL() { 550 this.memstore.startReplayingFromWAL(); 551 } 552 553 /** 554 * This message intends to inform the MemStore that the replaying edits from WAL are done 555 */ 556 public void stopReplayingFromWAL() { 557 this.memstore.stopReplayingFromWAL(); 558 } 559 560 /** 561 * Adds a value to the memstore 562 */ 563 public void add(final ExtendedCell cell, MemStoreSizing memstoreSizing) { 564 storeEngine.readLock(); 565 try { 566 if (this.currentParallelPutCount.getAndIncrement() > this.parallelPutCountPrintThreshold) { 567 LOG.trace("tableName={}, encodedName={}, columnFamilyName={} is too busy!", 568 this.getTableName(), this.getRegionInfo().getEncodedName(), this.getColumnFamilyName()); 569 } 570 this.memstore.add(cell, memstoreSizing); 571 } finally { 572 storeEngine.readUnlock(); 573 currentParallelPutCount.decrementAndGet(); 574 } 575 } 576 577 /** 578 * Adds the specified value to the memstore 579 */ 580 public void add(final Iterable<ExtendedCell> cells, MemStoreSizing memstoreSizing) { 581 storeEngine.readLock(); 582 try { 583 if (this.currentParallelPutCount.getAndIncrement() > this.parallelPutCountPrintThreshold) { 584 LOG.trace("tableName={}, encodedName={}, columnFamilyName={} is too busy!", 585 this.getTableName(), this.getRegionInfo().getEncodedName(), this.getColumnFamilyName()); 586 } 587 memstore.add(cells, memstoreSizing); 588 } finally { 589 storeEngine.readUnlock(); 590 currentParallelPutCount.decrementAndGet(); 591 } 592 } 593 594 @Override 595 public long timeOfOldestEdit() { 596 return memstore.timeOfOldestEdit(); 597 } 598 599 /** Returns All store files. */ 600 @Override 601 public Collection<HStoreFile> getStorefiles() { 602 return this.storeEngine.getStoreFileManager().getStoreFiles(); 603 } 604 605 @Override 606 public Collection<HStoreFile> getCompactedFiles() { 607 return this.storeEngine.getStoreFileManager().getCompactedfiles(); 608 } 609 610 /** 611 * This throws a WrongRegionException if the HFile does not fit in this region, or an 612 * InvalidHFileException if the HFile is not valid. 613 */ 614 public void assertBulkLoadHFileOk(Path srcPath) throws IOException { 615 HFile.Reader reader = null; 616 try { 617 LOG.info("Validating hfile at " + srcPath + " for inclusion in " + this); 618 FileSystem srcFs = srcPath.getFileSystem(conf); 619 srcFs.access(srcPath, FsAction.READ_WRITE); 620 reader = HFile.createReader(srcFs, srcPath, getCacheConfig(), isPrimaryReplicaStore(), conf); 621 622 Optional<byte[]> firstKey = reader.getFirstRowKey(); 623 Preconditions.checkState(firstKey.isPresent(), "First key can not be null"); 624 Optional<ExtendedCell> lk = reader.getLastKey(); 625 Preconditions.checkState(lk.isPresent(), "Last key can not be null"); 626 byte[] lastKey = CellUtil.cloneRow(lk.get()); 627 628 if (LOG.isDebugEnabled()) { 629 LOG.debug("HFile bounds: first=" + Bytes.toStringBinary(firstKey.get()) + " last=" 630 + Bytes.toStringBinary(lastKey)); 631 LOG.debug("Region bounds: first=" + Bytes.toStringBinary(getRegionInfo().getStartKey()) 632 + " last=" + Bytes.toStringBinary(getRegionInfo().getEndKey())); 633 } 634 635 if (!this.getRegionInfo().containsRange(firstKey.get(), lastKey)) { 636 throw new WrongRegionException("Bulk load file " + srcPath.toString() 637 + " does not fit inside region " + this.getRegionInfo().getRegionNameAsString()); 638 } 639 640 if ( 641 reader.length() 642 > conf.getLong(HConstants.HREGION_MAX_FILESIZE, HConstants.DEFAULT_MAX_FILE_SIZE) 643 ) { 644 LOG.warn("Trying to bulk load hfile " + srcPath + " with size: " + reader.length() 645 + " bytes can be problematic as it may lead to oversplitting."); 646 } 647 648 if (verifyBulkLoads) { 649 long verificationStartTime = EnvironmentEdgeManager.currentTime(); 650 LOG.info("Full verification started for bulk load hfile: {}", srcPath); 651 Cell prevCell = null; 652 HFileScanner scanner = reader.getScanner(conf, false, false, false); 653 scanner.seekTo(); 654 do { 655 Cell cell = scanner.getCell(); 656 if (prevCell != null) { 657 if (getComparator().compareRows(prevCell, cell) > 0) { 658 throw new InvalidHFileException("Previous row is greater than" + " current row: path=" 659 + srcPath + " previous=" + CellUtil.getCellKeyAsString(prevCell) + " current=" 660 + CellUtil.getCellKeyAsString(cell)); 661 } 662 if (CellComparator.getInstance().compareFamilies(prevCell, cell) != 0) { 663 throw new InvalidHFileException("Previous key had different" 664 + " family compared to current key: path=" + srcPath + " previous=" 665 + Bytes.toStringBinary(prevCell.getFamilyArray(), prevCell.getFamilyOffset(), 666 prevCell.getFamilyLength()) 667 + " current=" + Bytes.toStringBinary(cell.getFamilyArray(), cell.getFamilyOffset(), 668 cell.getFamilyLength())); 669 } 670 } 671 prevCell = cell; 672 } while (scanner.next()); 673 LOG.info("Full verification complete for bulk load hfile: " + srcPath.toString() + " took " 674 + (EnvironmentEdgeManager.currentTime() - verificationStartTime) + " ms"); 675 } 676 } finally { 677 if (reader != null) { 678 reader.close(); 679 } 680 } 681 } 682 683 /** 684 * This method should only be called from Region. It is assumed that the ranges of values in the 685 * HFile fit within the stores assigned region. (assertBulkLoadHFileOk checks this) 686 * @param seqNum sequence Id associated with the HFile 687 */ 688 public Pair<Path, Path> preBulkLoadHFile(String srcPathStr, long seqNum) throws IOException { 689 Path srcPath = new Path(srcPathStr); 690 return getRegionFileSystem().bulkLoadStoreFile(getColumnFamilyName(), srcPath, seqNum); 691 } 692 693 public Path bulkLoadHFile(byte[] family, String srcPathStr, Path dstPath) throws IOException { 694 Path srcPath = new Path(srcPathStr); 695 try { 696 getRegionFileSystem().commitStoreFile(srcPath, dstPath); 697 } finally { 698 if (this.getCoprocessorHost() != null) { 699 this.getCoprocessorHost().postCommitStoreFile(family, srcPath, dstPath); 700 } 701 } 702 703 LOG.info("Loaded HFile " + srcPath + " into " + this + " as " + dstPath 704 + " - updating store file list."); 705 706 HStoreFile sf = storeEngine.createStoreFileAndReader(dstPath); 707 bulkLoadHFile(sf); 708 709 LOG.info("Successfully loaded {} into {} (new location: {})", srcPath, this, dstPath); 710 711 return dstPath; 712 } 713 714 public void bulkLoadHFile(StoreFileInfo fileInfo) throws IOException { 715 HStoreFile sf = storeEngine.createStoreFileAndReader(fileInfo); 716 bulkLoadHFile(sf); 717 } 718 719 private void bulkLoadHFile(HStoreFile sf) throws IOException { 720 StoreFileReader r = sf.getReader(); 721 this.storeSize.addAndGet(r.length()); 722 this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes()); 723 storeEngine.addStoreFiles(Lists.newArrayList(sf), () -> { 724 }); 725 LOG.info("Loaded HFile " + sf.getFileInfo() + " into " + this); 726 if (LOG.isTraceEnabled()) { 727 String traceMessage = "BULK LOAD time,size,store size,store files [" 728 + EnvironmentEdgeManager.currentTime() + "," + r.length() + "," + storeSize + "," 729 + storeEngine.getStoreFileManager().getStorefileCount() + "]"; 730 LOG.trace(traceMessage); 731 } 732 } 733 734 private ImmutableCollection<HStoreFile> closeWithoutLock() throws IOException { 735 memstore.close(); 736 // Clear so metrics doesn't find them. 737 ImmutableCollection<HStoreFile> result = storeEngine.getStoreFileManager().clearFiles(); 738 Collection<HStoreFile> compactedfiles = storeEngine.getStoreFileManager().clearCompactedFiles(); 739 // clear the compacted files 740 if (CollectionUtils.isNotEmpty(compactedfiles)) { 741 removeCompactedfiles(compactedfiles, 742 getCacheConfig() != null ? getCacheConfig().shouldEvictOnClose() : true); 743 } 744 if (!result.isEmpty()) { 745 // initialize the thread pool for closing store files in parallel. 746 ThreadPoolExecutor storeFileCloserThreadPool = 747 this.region.getStoreFileOpenAndCloseThreadPool("StoreFileCloser-" 748 + this.region.getRegionInfo().getEncodedName() + "-" + this.getColumnFamilyName()); 749 750 // close each store file in parallel 751 CompletionService<Void> completionService = 752 new ExecutorCompletionService<>(storeFileCloserThreadPool); 753 for (HStoreFile f : result) { 754 completionService.submit(new Callable<Void>() { 755 @Override 756 public Void call() throws IOException { 757 boolean evictOnClose = 758 getCacheConfig() != null ? getCacheConfig().shouldEvictOnClose() : true; 759 f.closeStoreFile(!warmup && evictOnClose); 760 return null; 761 } 762 }); 763 } 764 765 IOException ioe = null; 766 try { 767 for (int i = 0; i < result.size(); i++) { 768 try { 769 Future<Void> future = completionService.take(); 770 future.get(); 771 } catch (InterruptedException e) { 772 if (ioe == null) { 773 ioe = new InterruptedIOException(); 774 ioe.initCause(e); 775 } 776 } catch (ExecutionException e) { 777 if (ioe == null) { 778 ioe = new IOException(e.getCause()); 779 } 780 } 781 } 782 } finally { 783 storeFileCloserThreadPool.shutdownNow(); 784 } 785 if (ioe != null) { 786 throw ioe; 787 } 788 } 789 LOG.trace("Closed {}", this); 790 return result; 791 } 792 793 /** 794 * Close all the readers We don't need to worry about subsequent requests because the Region holds 795 * a write lock that will prevent any more reads or writes. 796 * @return the {@link StoreFile StoreFiles} that were previously being used. 797 * @throws IOException on failure 798 */ 799 public ImmutableCollection<HStoreFile> close() throws IOException { 800 // findbugs can not recognize storeEngine.writeLock is just a lock operation so it will report 801 // UL_UNRELEASED_LOCK_EXCEPTION_PATH, so here we have to use two try finally... 802 // Change later if findbugs becomes smarter in the future. 803 this.archiveLock.lock(); 804 try { 805 this.storeEngine.writeLock(); 806 try { 807 return closeWithoutLock(); 808 } finally { 809 this.storeEngine.writeUnlock(); 810 } 811 } finally { 812 this.archiveLock.unlock(); 813 } 814 } 815 816 /** 817 * Write out current snapshot. Presumes {@code StoreFlusherImpl.prepare()} has been called 818 * previously. 819 * @param logCacheFlushId flush sequence number 820 * @return The path name of the tmp file to which the store was flushed 821 * @throws IOException if exception occurs during process 822 */ 823 protected List<Path> flushCache(final long logCacheFlushId, MemStoreSnapshot snapshot, 824 MonitoredTask status, ThroughputController throughputController, FlushLifeCycleTracker tracker, 825 Consumer<Path> writerCreationTracker) throws IOException { 826 // If an exception happens flushing, we let it out without clearing 827 // the memstore snapshot. The old snapshot will be returned when we say 828 // 'snapshot', the next time flush comes around. 829 // Retry after catching exception when flushing, otherwise server will abort 830 // itself 831 StoreFlusher flusher = storeEngine.getStoreFlusher(); 832 IOException lastException = null; 833 for (int i = 0; i < flushRetriesNumber; i++) { 834 try { 835 List<Path> pathNames = flusher.flushSnapshot(snapshot, logCacheFlushId, status, 836 throughputController, tracker, writerCreationTracker); 837 Path lastPathName = null; 838 try { 839 for (Path pathName : pathNames) { 840 lastPathName = pathName; 841 storeEngine.validateStoreFile(pathName, false); 842 } 843 return pathNames; 844 } catch (Exception e) { 845 LOG.warn("Failed validating store file {}, retrying num={}", lastPathName, i, e); 846 if (e instanceof IOException) { 847 lastException = (IOException) e; 848 } else { 849 lastException = new IOException(e); 850 } 851 } 852 } catch (IOException e) { 853 LOG.warn("Failed flushing store file for {}, retrying num={}", this, i, e); 854 lastException = e; 855 } 856 if (lastException != null && i < (flushRetriesNumber - 1)) { 857 try { 858 Thread.sleep(pauseTime); 859 } catch (InterruptedException e) { 860 IOException iie = new InterruptedIOException(); 861 iie.initCause(e); 862 throw iie; 863 } 864 } 865 } 866 throw lastException; 867 } 868 869 public HStoreFile tryCommitRecoveredHFile(Path path) throws IOException { 870 LOG.info("Validating recovered hfile at {} for inclusion in store {}", path, this); 871 FileSystem srcFs = path.getFileSystem(conf); 872 srcFs.access(path, FsAction.READ_WRITE); 873 try (HFile.Reader reader = 874 HFile.createReader(srcFs, path, getCacheConfig(), isPrimaryReplicaStore(), conf)) { 875 Optional<byte[]> firstKey = reader.getFirstRowKey(); 876 Preconditions.checkState(firstKey.isPresent(), "First key can not be null"); 877 Optional<ExtendedCell> lk = reader.getLastKey(); 878 Preconditions.checkState(lk.isPresent(), "Last key can not be null"); 879 byte[] lastKey = CellUtil.cloneRow(lk.get()); 880 if (!this.getRegionInfo().containsRange(firstKey.get(), lastKey)) { 881 throw new WrongRegionException("Recovered hfile " + path.toString() 882 + " does not fit inside region " + this.getRegionInfo().getRegionNameAsString()); 883 } 884 } 885 886 Path dstPath = getRegionFileSystem().commitStoreFile(getColumnFamilyName(), path); 887 HStoreFile sf = storeEngine.createStoreFileAndReader(dstPath); 888 StoreFileReader r = sf.getReader(); 889 this.storeSize.addAndGet(r.length()); 890 this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes()); 891 892 storeEngine.addStoreFiles(Lists.newArrayList(sf), () -> { 893 }); 894 895 LOG.info("Loaded recovered hfile to {}, entries={}, sequenceid={}, filesize={}", sf, 896 r.getEntries(), r.getSequenceID(), TraditionalBinaryPrefix.long2String(r.length(), "B", 1)); 897 return sf; 898 } 899 900 private long getTotalSize(Collection<HStoreFile> sfs) { 901 return sfs.stream().mapToLong(sf -> sf.getReader().length()).sum(); 902 } 903 904 private boolean completeFlush(final List<HStoreFile> sfs, long snapshotId) throws IOException { 905 // NOTE:we should keep clearSnapshot method inside the write lock because clearSnapshot may 906 // close {@link DefaultMemStore#snapshot}, which may be used by 907 // {@link DefaultMemStore#getScanners}. 908 storeEngine.addStoreFiles(sfs, 909 // NOTE: here we must increase the refCount for storeFiles because we would open the 910 // storeFiles and get the StoreFileScanners for them in HStore.notifyChangedReadersObservers. 911 // If we don't increase the refCount here, HStore.closeAndArchiveCompactedFiles called by 912 // CompactedHFilesDischarger may archive the storeFiles after a concurrent compaction.Because 913 // HStore.requestCompaction is under storeEngine lock, so here we increase the refCount under 914 // storeEngine lock. see HBASE-27519 for more details. 915 snapshotId > 0 ? () -> { 916 this.memstore.clearSnapshot(snapshotId); 917 HStoreFile.increaseStoreFilesRefeCount(sfs); 918 } : () -> { 919 HStoreFile.increaseStoreFilesRefeCount(sfs); 920 }); 921 // notify to be called here - only in case of flushes 922 try { 923 notifyChangedReadersObservers(sfs); 924 } finally { 925 HStoreFile.decreaseStoreFilesRefeCount(sfs); 926 } 927 if (LOG.isTraceEnabled()) { 928 long totalSize = getTotalSize(sfs); 929 String traceMessage = "FLUSH time,count,size,store size,store files [" 930 + EnvironmentEdgeManager.currentTime() + "," + sfs.size() + "," + totalSize + "," 931 + storeSize + "," + storeEngine.getStoreFileManager().getStorefileCount() + "]"; 932 LOG.trace(traceMessage); 933 } 934 return needsCompaction(); 935 } 936 937 /** 938 * Notify all observers that set of Readers has changed. 939 */ 940 private void notifyChangedReadersObservers(List<HStoreFile> sfs) throws IOException { 941 for (ChangedReadersObserver o : this.changedReaderObservers) { 942 List<KeyValueScanner> memStoreScanners; 943 this.storeEngine.readLock(); 944 try { 945 memStoreScanners = this.memstore.getScanners(o.getReadPoint()); 946 } finally { 947 this.storeEngine.readUnlock(); 948 } 949 o.updateReaders(sfs, memStoreScanners); 950 } 951 } 952 953 /** 954 * Get all scanners with no filtering based on TTL (that happens further down the line). 955 * @param cacheBlocks cache the blocks or not 956 * @param usePread true to use pread, false if not 957 * @param isCompaction true if the scanner is created for compaction 958 * @param matcher the scan query matcher 959 * @param startRow the start row 960 * @param stopRow the stop row 961 * @param readPt the read point of the current scan 962 * @return all scanners for this store 963 */ 964 public List<KeyValueScanner> getScanners(boolean cacheBlocks, boolean isGet, boolean usePread, 965 boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow, byte[] stopRow, long readPt, 966 boolean onlyLatestVersion) throws IOException { 967 return getScanners(cacheBlocks, usePread, isCompaction, matcher, startRow, true, stopRow, false, 968 readPt, onlyLatestVersion); 969 } 970 971 /** 972 * Get all scanners with no filtering based on TTL (that happens further down the line). 973 * @param cacheBlocks cache the blocks or not 974 * @param usePread true to use pread, false if not 975 * @param isCompaction true if the scanner is created for compaction 976 * @param matcher the scan query matcher 977 * @param startRow the start row 978 * @param includeStartRow true to include start row, false if not 979 * @param stopRow the stop row 980 * @param includeStopRow true to include stop row, false if not 981 * @param readPt the read point of the current scan 982 * @return all scanners for this store 983 */ 984 public List<KeyValueScanner> getScanners(boolean cacheBlocks, boolean usePread, 985 boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow, boolean includeStartRow, 986 byte[] stopRow, boolean includeStopRow, long readPt, boolean onlyLatestVersion) 987 throws IOException { 988 Collection<HStoreFile> storeFilesToScan; 989 List<KeyValueScanner> memStoreScanners; 990 this.storeEngine.readLock(); 991 try { 992 storeFilesToScan = this.storeEngine.getStoreFileManager().getFilesForScan(startRow, 993 includeStartRow, stopRow, includeStopRow, onlyLatestVersion); 994 memStoreScanners = this.memstore.getScanners(readPt); 995 // NOTE: here we must increase the refCount for storeFiles because we would open the 996 // storeFiles and get the StoreFileScanners for them.If we don't increase the refCount here, 997 // HStore.closeAndArchiveCompactedFiles called by CompactedHFilesDischarger may archive the 998 // storeFiles after a concurrent compaction.Because HStore.requestCompaction is under 999 // storeEngine lock, so here we increase the refCount under storeEngine lock. see HBASE-27484 1000 // for more details. 1001 HStoreFile.increaseStoreFilesRefeCount(storeFilesToScan); 1002 } finally { 1003 this.storeEngine.readUnlock(); 1004 } 1005 try { 1006 // First the store file scanners 1007 1008 // TODO this used to get the store files in descending order, 1009 // but now we get them in ascending order, which I think is 1010 // actually more correct, since memstore get put at the end. 1011 List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles( 1012 storeFilesToScan, cacheBlocks, usePread, isCompaction, false, matcher, readPt); 1013 List<KeyValueScanner> scanners = new ArrayList<>(sfScanners.size() + 1); 1014 scanners.addAll(sfScanners); 1015 // Then the memstore scanners 1016 scanners.addAll(memStoreScanners); 1017 return scanners; 1018 } catch (Throwable t) { 1019 clearAndClose(memStoreScanners); 1020 throw t instanceof IOException ? (IOException) t : new IOException(t); 1021 } finally { 1022 HStoreFile.decreaseStoreFilesRefeCount(storeFilesToScan); 1023 } 1024 } 1025 1026 private static void clearAndClose(List<KeyValueScanner> scanners) { 1027 if (scanners == null) { 1028 return; 1029 } 1030 for (KeyValueScanner s : scanners) { 1031 s.close(); 1032 } 1033 scanners.clear(); 1034 } 1035 1036 /** 1037 * Create scanners on the given files and if needed on the memstore with no filtering based on TTL 1038 * (that happens further down the line). 1039 * @param files the list of files on which the scanners has to be created 1040 * @param cacheBlocks cache the blocks or not 1041 * @param usePread true to use pread, false if not 1042 * @param isCompaction true if the scanner is created for compaction 1043 * @param matcher the scan query matcher 1044 * @param startRow the start row 1045 * @param stopRow the stop row 1046 * @param readPt the read point of the current scan 1047 * @param includeMemstoreScanner true if memstore has to be included 1048 * @return scanners on the given files and on the memstore if specified 1049 */ 1050 public List<KeyValueScanner> getScanners(List<HStoreFile> files, boolean cacheBlocks, 1051 boolean isGet, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher, 1052 byte[] startRow, byte[] stopRow, long readPt, boolean includeMemstoreScanner, 1053 boolean onlyLatestVersion) throws IOException { 1054 return getScanners(files, cacheBlocks, usePread, isCompaction, matcher, startRow, true, stopRow, 1055 false, readPt, includeMemstoreScanner, onlyLatestVersion); 1056 } 1057 1058 /** 1059 * Create scanners on the given files and if needed on the memstore with no filtering based on TTL 1060 * (that happens further down the line). 1061 * @param files the list of files on which the scanners has to be created 1062 * @param cacheBlocks ache the blocks or not 1063 * @param usePread true to use pread, false if not 1064 * @param isCompaction true if the scanner is created for compaction 1065 * @param matcher the scan query matcher 1066 * @param startRow the start row 1067 * @param includeStartRow true to include start row, false if not 1068 * @param stopRow the stop row 1069 * @param includeStopRow true to include stop row, false if not 1070 * @param readPt the read point of the current scan 1071 * @param includeMemstoreScanner true if memstore has to be included 1072 * @return scanners on the given files and on the memstore if specified 1073 */ 1074 public List<KeyValueScanner> getScanners(List<HStoreFile> files, boolean cacheBlocks, 1075 boolean usePread, boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow, 1076 boolean includeStartRow, byte[] stopRow, boolean includeStopRow, long readPt, 1077 boolean includeMemstoreScanner, boolean onlyLatestVersion) throws IOException { 1078 List<KeyValueScanner> memStoreScanners = null; 1079 if (includeMemstoreScanner) { 1080 this.storeEngine.readLock(); 1081 try { 1082 memStoreScanners = this.memstore.getScanners(readPt); 1083 } finally { 1084 this.storeEngine.readUnlock(); 1085 } 1086 } 1087 try { 1088 List<StoreFileScanner> sfScanners = StoreFileScanner.getScannersForStoreFiles(files, 1089 cacheBlocks, usePread, isCompaction, false, matcher, readPt); 1090 List<KeyValueScanner> scanners = new ArrayList<>(sfScanners.size() + 1); 1091 scanners.addAll(sfScanners); 1092 // Then the memstore scanners 1093 if (memStoreScanners != null) { 1094 scanners.addAll(memStoreScanners); 1095 } 1096 return scanners; 1097 } catch (Throwable t) { 1098 clearAndClose(memStoreScanners); 1099 throw t instanceof IOException ? (IOException) t : new IOException(t); 1100 } 1101 } 1102 1103 /** 1104 * @param o Observer who wants to know about changes in set of Readers 1105 */ 1106 public void addChangedReaderObserver(ChangedReadersObserver o) { 1107 this.changedReaderObservers.add(o); 1108 } 1109 1110 /** 1111 * @param o Observer no longer interested in changes in set of Readers. 1112 */ 1113 public void deleteChangedReaderObserver(ChangedReadersObserver o) { 1114 // We don't check if observer present; it may not be (legitimately) 1115 this.changedReaderObservers.remove(o); 1116 } 1117 1118 ////////////////////////////////////////////////////////////////////////////// 1119 // Compaction 1120 ////////////////////////////////////////////////////////////////////////////// 1121 1122 /** 1123 * Compact the StoreFiles. This method may take some time, so the calling thread must be able to 1124 * block for long periods. 1125 * <p> 1126 * During this time, the Store can work as usual, getting values from StoreFiles and writing new 1127 * StoreFiles from the MemStore. Existing StoreFiles are not destroyed until the new compacted 1128 * StoreFile is completely written-out to disk. 1129 * <p> 1130 * The compactLock prevents multiple simultaneous compactions. The structureLock prevents us from 1131 * interfering with other write operations. 1132 * <p> 1133 * We don't want to hold the structureLock for the whole time, as a compact() can be lengthy and 1134 * we want to allow cache-flushes during this period. 1135 * <p> 1136 * Compaction event should be idempotent, since there is no IO Fencing for the region directory in 1137 * hdfs. A region server might still try to complete the compaction after it lost the region. That 1138 * is why the following events are carefully ordered for a compaction: 1139 * <ol> 1140 * <li>Compaction writes new files under region/.tmp directory (compaction output)</li> 1141 * <li>Compaction atomically moves the temporary file under region directory</li> 1142 * <li>Compaction appends a WAL edit containing the compaction input and output files. Forces sync 1143 * on WAL.</li> 1144 * <li>Compaction deletes the input files from the region directory.</li> 1145 * </ol> 1146 * Failure conditions are handled like this: 1147 * <ul> 1148 * <li>If RS fails before 2, compaction won't complete. Even if RS lives on and finishes the 1149 * compaction later, it will only write the new data file to the region directory. Since we 1150 * already have this data, this will be idempotent, but we will have a redundant copy of the 1151 * data.</li> 1152 * <li>If RS fails between 2 and 3, the region will have a redundant copy of the data. The RS that 1153 * failed won't be able to finish sync() for WAL because of lease recovery in WAL.</li> 1154 * <li>If RS fails after 3, the region server who opens the region will pick up the compaction 1155 * marker from the WAL and replay it by removing the compaction input files. Failed RS can also 1156 * attempt to delete those files, but the operation will be idempotent</li> 1157 * </ul> 1158 * See HBASE-2231 for details. 1159 * @param compaction compaction details obtained from requestCompaction() 1160 * @return The storefiles that we compacted into or null if we failed or opted out early. 1161 */ 1162 public List<HStoreFile> compact(CompactionContext compaction, 1163 ThroughputController throughputController, User user) throws IOException { 1164 assert compaction != null; 1165 CompactionRequestImpl cr = compaction.getRequest(); 1166 StoreFileWriterCreationTracker writerCreationTracker = 1167 storeFileWriterCreationTrackerFactory.get(); 1168 if (writerCreationTracker != null) { 1169 cr.setWriterCreationTracker(writerCreationTracker); 1170 storeFileWriterCreationTrackers.add(writerCreationTracker); 1171 } 1172 try { 1173 // Do all sanity checking in here if we have a valid CompactionRequestImpl 1174 // because we need to clean up after it on the way out in a finally 1175 // block below 1176 long compactionStartTime = EnvironmentEdgeManager.currentTime(); 1177 assert compaction.hasSelection(); 1178 Collection<HStoreFile> filesToCompact = cr.getFiles(); 1179 assert !filesToCompact.isEmpty(); 1180 synchronized (filesCompacting) { 1181 // sanity check: we're compacting files that this store knows about 1182 // TODO: change this to LOG.error() after more debugging 1183 Preconditions.checkArgument(filesCompacting.containsAll(filesToCompact)); 1184 } 1185 1186 // Ready to go. Have list of files to compact. 1187 LOG.info("Starting compaction of " + filesToCompact + " into tmpdir=" 1188 + getRegionFileSystem().getTempDir() + ", totalSize=" 1189 + TraditionalBinaryPrefix.long2String(cr.getSize(), "", 1)); 1190 1191 return doCompaction(cr, filesToCompact, user, compactionStartTime, 1192 compaction.compact(throughputController, user)); 1193 } finally { 1194 finishCompactionRequest(cr); 1195 } 1196 } 1197 1198 protected List<HStoreFile> doCompaction(CompactionRequestImpl cr, 1199 Collection<HStoreFile> filesToCompact, User user, long compactionStartTime, List<Path> newFiles) 1200 throws IOException { 1201 // Do the steps necessary to complete the compaction. 1202 setStoragePolicyFromFileName(newFiles); 1203 List<HStoreFile> sfs = storeEngine.commitStoreFiles(newFiles, true, true); 1204 if (this.getCoprocessorHost() != null) { 1205 for (HStoreFile sf : sfs) { 1206 getCoprocessorHost().postCompact(this, sf, cr.getTracker(), cr, user); 1207 } 1208 } 1209 replaceStoreFiles(filesToCompact, sfs, true); 1210 1211 long outputBytes = getTotalSize(sfs); 1212 1213 // At this point the store will use new files for all new scanners. 1214 refreshStoreSizeAndTotalBytes(); // update store size. 1215 1216 long now = EnvironmentEdgeManager.currentTime(); 1217 if ( 1218 region.getRegionServerServices() != null 1219 && region.getRegionServerServices().getMetrics() != null 1220 ) { 1221 region.getRegionServerServices().getMetrics().updateCompaction( 1222 region.getTableDescriptor().getTableName().getNameAsString(), cr.isMajor(), 1223 now - compactionStartTime, cr.getFiles().size(), newFiles.size(), cr.getSize(), 1224 outputBytes); 1225 1226 } 1227 1228 logCompactionEndMessage(cr, sfs, now, compactionStartTime); 1229 return sfs; 1230 } 1231 1232 // Set correct storage policy from the file name of DTCP. 1233 // Rename file will not change the storage policy. 1234 private void setStoragePolicyFromFileName(List<Path> newFiles) throws IOException { 1235 String prefix = HConstants.STORAGE_POLICY_PREFIX; 1236 for (Path newFile : newFiles) { 1237 if (newFile.getParent().getName().startsWith(prefix)) { 1238 CommonFSUtils.setStoragePolicy(getRegionFileSystem().getFileSystem(), newFile, 1239 newFile.getParent().getName().substring(prefix.length())); 1240 } 1241 } 1242 } 1243 1244 /** 1245 * Writes the compaction WAL record. 1246 * @param filesCompacted Files compacted (input). 1247 * @param newFiles Files from compaction. 1248 */ 1249 private void writeCompactionWalRecord(Collection<HStoreFile> filesCompacted, 1250 Collection<HStoreFile> newFiles) throws IOException { 1251 if (region.getWAL() == null) { 1252 return; 1253 } 1254 List<Path> inputPaths = 1255 filesCompacted.stream().map(HStoreFile::getPath).collect(Collectors.toList()); 1256 List<Path> outputPaths = 1257 newFiles.stream().map(HStoreFile::getPath).collect(Collectors.toList()); 1258 RegionInfo info = this.region.getRegionInfo(); 1259 CompactionDescriptor compactionDescriptor = ProtobufUtil.toCompactionDescriptor(info, 1260 getColumnFamilyDescriptor().getName(), inputPaths, outputPaths, 1261 getRegionFileSystem().getStoreDir(getColumnFamilyDescriptor().getNameAsString())); 1262 // Fix reaching into Region to get the maxWaitForSeqId. 1263 // Does this method belong in Region altogether given it is making so many references up there? 1264 // Could be Region#writeCompactionMarker(compactionDescriptor); 1265 WALUtil.writeCompactionMarker(this.region.getWAL(), this.region.getReplicationScope(), 1266 this.region.getRegionInfo(), compactionDescriptor, this.region.getMVCC(), 1267 region.getRegionReplicationSink().orElse(null)); 1268 } 1269 1270 @RestrictedApi(explanation = "Should only be called in TestHStore", link = "", 1271 allowedOnPath = ".*/(HStore|TestHStore).java") 1272 void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result, 1273 boolean writeCompactionMarker) throws IOException { 1274 storeEngine.replaceStoreFiles(compactedFiles, result, () -> { 1275 if (writeCompactionMarker) { 1276 writeCompactionWalRecord(compactedFiles, result); 1277 } 1278 }, () -> { 1279 synchronized (filesCompacting) { 1280 filesCompacting.removeAll(compactedFiles); 1281 } 1282 }); 1283 // These may be null when the RS is shutting down. The space quota Chores will fix the Region 1284 // sizes later so it's not super-critical if we miss these. 1285 RegionServerServices rsServices = region.getRegionServerServices(); 1286 if (rsServices != null && rsServices.getRegionServerSpaceQuotaManager() != null) { 1287 updateSpaceQuotaAfterFileReplacement( 1288 rsServices.getRegionServerSpaceQuotaManager().getRegionSizeStore(), getRegionInfo(), 1289 compactedFiles, result); 1290 } 1291 } 1292 1293 /** 1294 * Updates the space quota usage for this region, removing the size for files compacted away and 1295 * adding in the size for new files. 1296 * @param sizeStore The object tracking changes in region size for space quotas. 1297 * @param regionInfo The identifier for the region whose size is being updated. 1298 * @param oldFiles Files removed from this store's region. 1299 * @param newFiles Files added to this store's region. 1300 */ 1301 void updateSpaceQuotaAfterFileReplacement(RegionSizeStore sizeStore, RegionInfo regionInfo, 1302 Collection<HStoreFile> oldFiles, Collection<HStoreFile> newFiles) { 1303 long delta = 0; 1304 if (oldFiles != null) { 1305 for (HStoreFile compactedFile : oldFiles) { 1306 if (compactedFile.isHFile()) { 1307 delta -= compactedFile.getReader().length(); 1308 } 1309 } 1310 } 1311 if (newFiles != null) { 1312 for (HStoreFile newFile : newFiles) { 1313 if (newFile.isHFile()) { 1314 delta += newFile.getReader().length(); 1315 } 1316 } 1317 } 1318 sizeStore.incrementRegionSize(regionInfo, delta); 1319 } 1320 1321 /** 1322 * Log a very elaborate compaction completion message. 1323 * @param cr Request. 1324 * @param sfs Resulting files. 1325 * @param compactionStartTime Start time. 1326 */ 1327 private void logCompactionEndMessage(CompactionRequestImpl cr, List<HStoreFile> sfs, long now, 1328 long compactionStartTime) { 1329 StringBuilder message = new StringBuilder("Completed" + (cr.isMajor() ? " major" : "") 1330 + " compaction of " + cr.getFiles().size() + (cr.isAllFiles() ? " (all)" : "") 1331 + " file(s) in " + this + " of " + this.getRegionInfo().getShortNameToLog() + " into "); 1332 if (sfs.isEmpty()) { 1333 message.append("none, "); 1334 } else { 1335 for (HStoreFile sf : sfs) { 1336 message.append(sf.getPath().getName()); 1337 message.append("(size="); 1338 message.append(TraditionalBinaryPrefix.long2String(sf.getReader().length(), "", 1)); 1339 message.append("), "); 1340 } 1341 } 1342 message.append("total size for store is ") 1343 .append(StringUtils.TraditionalBinaryPrefix.long2String(storeSize.get(), "", 1)) 1344 .append(". This selection was in queue for ") 1345 .append(StringUtils.formatTimeDiff(compactionStartTime, cr.getSelectionTime())) 1346 .append(", and took ").append(StringUtils.formatTimeDiff(now, compactionStartTime)) 1347 .append(" to execute."); 1348 LOG.info(message.toString()); 1349 if (LOG.isTraceEnabled()) { 1350 int fileCount = storeEngine.getStoreFileManager().getStorefileCount(); 1351 long resultSize = getTotalSize(sfs); 1352 String traceMessage = "COMPACTION start,end,size out,files in,files out,store size," 1353 + "store files [" + compactionStartTime + "," + now + "," + resultSize + "," 1354 + cr.getFiles().size() + "," + sfs.size() + "," + storeSize + "," + fileCount + "]"; 1355 LOG.trace(traceMessage); 1356 } 1357 } 1358 1359 /** 1360 * Call to complete a compaction. Its for the case where we find in the WAL a compaction that was 1361 * not finished. We could find one recovering a WAL after a regionserver crash. See HBASE-2231. 1362 */ 1363 public void replayCompactionMarker(CompactionDescriptor compaction, boolean pickCompactionFiles, 1364 boolean removeFiles) throws IOException { 1365 LOG.debug("Completing compaction from the WAL marker"); 1366 List<String> compactionInputs = compaction.getCompactionInputList(); 1367 List<String> compactionOutputs = Lists.newArrayList(compaction.getCompactionOutputList()); 1368 1369 // The Compaction Marker is written after the compaction is completed, 1370 // and the files moved into the region/family folder. 1371 // 1372 // If we crash after the entry is written, we may not have removed the 1373 // input files, but the output file is present. 1374 // (The unremoved input files will be removed by this function) 1375 // 1376 // If we scan the directory and the file is not present, it can mean that: 1377 // - The file was manually removed by the user 1378 // - The file was removed as consequence of subsequent compaction 1379 // so, we can't do anything with the "compaction output list" because those 1380 // files have already been loaded when opening the region (by virtue of 1381 // being in the store's folder) or they may be missing due to a compaction. 1382 1383 String familyName = this.getColumnFamilyName(); 1384 Set<String> inputFiles = new HashSet<>(); 1385 for (String compactionInput : compactionInputs) { 1386 Path inputPath = getRegionFileSystem().getStoreFilePath(familyName, compactionInput); 1387 inputFiles.add(inputPath.getName()); 1388 } 1389 1390 // some of the input files might already be deleted 1391 List<HStoreFile> inputStoreFiles = new ArrayList<>(compactionInputs.size()); 1392 for (HStoreFile sf : this.getStorefiles()) { 1393 if (inputFiles.contains(sf.getPath().getName())) { 1394 inputStoreFiles.add(sf); 1395 } 1396 } 1397 1398 // check whether we need to pick up the new files 1399 List<HStoreFile> outputStoreFiles = new ArrayList<>(compactionOutputs.size()); 1400 1401 if (pickCompactionFiles) { 1402 for (HStoreFile sf : this.getStorefiles()) { 1403 compactionOutputs.remove(sf.getPath().getName()); 1404 } 1405 for (String compactionOutput : compactionOutputs) { 1406 StoreFileTracker sft = StoreFileTrackerFactory.create(conf, false, storeContext); 1407 StoreFileInfo storeFileInfo = 1408 getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), compactionOutput, sft); 1409 HStoreFile storeFile = storeEngine.createStoreFileAndReader(storeFileInfo); 1410 outputStoreFiles.add(storeFile); 1411 } 1412 } 1413 1414 if (!inputStoreFiles.isEmpty() || !outputStoreFiles.isEmpty()) { 1415 LOG.info("Replaying compaction marker, replacing input files: " + inputStoreFiles 1416 + " with output files : " + outputStoreFiles); 1417 this.replaceStoreFiles(inputStoreFiles, outputStoreFiles, false); 1418 this.refreshStoreSizeAndTotalBytes(); 1419 } 1420 } 1421 1422 @Override 1423 public boolean hasReferences() { 1424 // Grab the read lock here, because we need to ensure that: only when the atomic 1425 // replaceStoreFiles(..) finished, we can get all the complete store file list. 1426 this.storeEngine.readLock(); 1427 try { 1428 // Merge the current store files with compacted files here due to HBASE-20940. 1429 Collection<HStoreFile> allStoreFiles = new ArrayList<>(getStorefiles()); 1430 allStoreFiles.addAll(getCompactedFiles()); 1431 return StoreUtils.hasReferences(allStoreFiles); 1432 } finally { 1433 this.storeEngine.readUnlock(); 1434 } 1435 } 1436 1437 /** 1438 * getter for CompactionProgress object 1439 * @return CompactionProgress object; can be null 1440 */ 1441 public CompactionProgress getCompactionProgress() { 1442 return this.storeEngine.getCompactor().getProgress(); 1443 } 1444 1445 @Override 1446 public boolean shouldPerformMajorCompaction() throws IOException { 1447 for (HStoreFile sf : this.storeEngine.getStoreFileManager().getStoreFiles()) { 1448 // TODO: what are these reader checks all over the place? 1449 if (sf.getReader() == null) { 1450 LOG.debug("StoreFile {} has null Reader", sf); 1451 return false; 1452 } 1453 } 1454 return storeEngine.getCompactionPolicy() 1455 .shouldPerformMajorCompaction(this.storeEngine.getStoreFileManager().getStoreFiles()); 1456 } 1457 1458 public Optional<CompactionContext> requestCompaction() throws IOException { 1459 return requestCompaction(NO_PRIORITY, CompactionLifeCycleTracker.DUMMY, null); 1460 } 1461 1462 public Optional<CompactionContext> requestCompaction(int priority, 1463 CompactionLifeCycleTracker tracker, User user) throws IOException { 1464 // don't even select for compaction if writes are disabled 1465 if (!this.areWritesEnabled()) { 1466 return Optional.empty(); 1467 } 1468 // Before we do compaction, try to get rid of unneeded files to simplify things. 1469 removeUnneededFiles(); 1470 1471 final CompactionContext compaction = storeEngine.createCompaction(); 1472 CompactionRequestImpl request = null; 1473 this.storeEngine.readLock(); 1474 try { 1475 synchronized (filesCompacting) { 1476 // First, see if coprocessor would want to override selection. 1477 if (this.getCoprocessorHost() != null) { 1478 final List<HStoreFile> candidatesForCoproc = compaction.preSelect(this.filesCompacting); 1479 boolean override = 1480 getCoprocessorHost().preCompactSelection(this, candidatesForCoproc, tracker, user); 1481 if (override) { 1482 // Coprocessor is overriding normal file selection. 1483 compaction.forceSelect(new CompactionRequestImpl(candidatesForCoproc)); 1484 } 1485 } 1486 1487 // Normal case - coprocessor is not overriding file selection. 1488 if (!compaction.hasSelection()) { 1489 boolean isUserCompaction = priority == Store.PRIORITY_USER; 1490 boolean mayUseOffPeak = 1491 offPeakHours.isOffPeakHour() && offPeakCompactionTracker.compareAndSet(false, true); 1492 try { 1493 compaction.select(this.filesCompacting, isUserCompaction, mayUseOffPeak, 1494 forceMajor && filesCompacting.isEmpty()); 1495 } catch (IOException e) { 1496 if (mayUseOffPeak) { 1497 offPeakCompactionTracker.set(false); 1498 } 1499 throw e; 1500 } 1501 assert compaction.hasSelection(); 1502 if (mayUseOffPeak && !compaction.getRequest().isOffPeak()) { 1503 // Compaction policy doesn't want to take advantage of off-peak. 1504 offPeakCompactionTracker.set(false); 1505 } 1506 } 1507 if (this.getCoprocessorHost() != null) { 1508 this.getCoprocessorHost().postCompactSelection(this, 1509 ImmutableList.copyOf(compaction.getRequest().getFiles()), tracker, 1510 compaction.getRequest(), user); 1511 } 1512 // Finally, we have the resulting files list. Check if we have any files at all. 1513 request = compaction.getRequest(); 1514 Collection<HStoreFile> selectedFiles = request.getFiles(); 1515 if (selectedFiles.isEmpty()) { 1516 return Optional.empty(); 1517 } 1518 1519 addToCompactingFiles(selectedFiles); 1520 1521 // If we're enqueuing a major, clear the force flag. 1522 this.forceMajor = this.forceMajor && !request.isMajor(); 1523 1524 // Set common request properties. 1525 // Set priority, either override value supplied by caller or from store. 1526 final int compactionPriority = 1527 (priority != Store.NO_PRIORITY) ? priority : getCompactPriority(); 1528 request.setPriority(compactionPriority); 1529 1530 if (request.isAfterSplit()) { 1531 // If the store belongs to recently splitted daughter regions, better we consider 1532 // them with the higher priority in the compaction queue. 1533 // Override priority if it is lower (higher int value) than 1534 // SPLIT_REGION_COMPACTION_PRIORITY 1535 final int splitHousekeepingPriority = 1536 Math.min(compactionPriority, SPLIT_REGION_COMPACTION_PRIORITY); 1537 request.setPriority(splitHousekeepingPriority); 1538 LOG.info( 1539 "Keeping/Overriding Compaction request priority to {} for CF {} since it" 1540 + " belongs to recently split daughter region {}", 1541 splitHousekeepingPriority, this.getColumnFamilyName(), 1542 getRegionInfo().getRegionNameAsString()); 1543 } 1544 request.setDescription(getRegionInfo().getRegionNameAsString(), getColumnFamilyName()); 1545 request.setTracker(tracker); 1546 } 1547 } finally { 1548 this.storeEngine.readUnlock(); 1549 } 1550 1551 if (LOG.isDebugEnabled()) { 1552 LOG.debug(this + " is initiating " + (request.isMajor() ? "major" : "minor") + " compaction" 1553 + (request.isAllFiles() ? " (all files)" : "")); 1554 } 1555 this.region.reportCompactionRequestStart(request.isMajor()); 1556 return Optional.of(compaction); 1557 } 1558 1559 /** Adds the files to compacting files. filesCompacting must be locked. */ 1560 private void addToCompactingFiles(Collection<HStoreFile> filesToAdd) { 1561 if (CollectionUtils.isEmpty(filesToAdd)) { 1562 return; 1563 } 1564 // Check that we do not try to compact the same StoreFile twice. 1565 if (!Collections.disjoint(filesCompacting, filesToAdd)) { 1566 Preconditions.checkArgument(false, "%s overlaps with %s", filesToAdd, filesCompacting); 1567 } 1568 filesCompacting.addAll(filesToAdd); 1569 Collections.sort(filesCompacting, storeEngine.getStoreFileManager().getStoreFileComparator()); 1570 } 1571 1572 private void removeUnneededFiles() throws IOException { 1573 if (!conf.getBoolean("hbase.store.delete.expired.storefile", true)) { 1574 return; 1575 } 1576 if (getColumnFamilyDescriptor().getMinVersions() > 0) { 1577 LOG.debug("Skipping expired store file removal due to min version of {} being {}", this, 1578 getColumnFamilyDescriptor().getMinVersions()); 1579 return; 1580 } 1581 this.storeEngine.readLock(); 1582 Collection<HStoreFile> delSfs = null; 1583 try { 1584 synchronized (filesCompacting) { 1585 long cfTtl = getStoreFileTtl(); 1586 if (cfTtl != Long.MAX_VALUE) { 1587 delSfs = storeEngine.getStoreFileManager() 1588 .getUnneededFiles(EnvironmentEdgeManager.currentTime() - cfTtl, filesCompacting); 1589 addToCompactingFiles(delSfs); 1590 } 1591 } 1592 } finally { 1593 this.storeEngine.readUnlock(); 1594 } 1595 1596 if (CollectionUtils.isEmpty(delSfs)) { 1597 return; 1598 } 1599 1600 Collection<HStoreFile> newFiles = Collections.emptyList(); // No new files. 1601 replaceStoreFiles(delSfs, newFiles, true); 1602 refreshStoreSizeAndTotalBytes(); 1603 LOG.info("Completed removal of " + delSfs.size() + " unnecessary (expired) file(s) in " + this 1604 + "; total size is " + TraditionalBinaryPrefix.long2String(storeSize.get(), "", 1)); 1605 } 1606 1607 public void cancelRequestedCompaction(CompactionContext compaction) { 1608 finishCompactionRequest(compaction.getRequest()); 1609 } 1610 1611 private void finishCompactionRequest(CompactionRequestImpl cr) { 1612 this.region.reportCompactionRequestEnd(cr.isMajor(), cr.getFiles().size(), cr.getSize()); 1613 if (cr.isOffPeak()) { 1614 offPeakCompactionTracker.set(false); 1615 cr.setOffPeak(false); 1616 } 1617 synchronized (filesCompacting) { 1618 filesCompacting.removeAll(cr.getFiles()); 1619 } 1620 // The tracker could be null, for example, we do not need to track the creation of store file 1621 // writer due to different implementation of SFT, or the compaction is canceled. 1622 if (cr.getWriterCreationTracker() != null) { 1623 storeFileWriterCreationTrackers.remove(cr.getWriterCreationTracker()); 1624 } 1625 } 1626 1627 /** 1628 * Update counts. 1629 */ 1630 protected void refreshStoreSizeAndTotalBytes() throws IOException { 1631 this.storeSize.set(0L); 1632 this.totalUncompressedBytes.set(0L); 1633 for (HStoreFile hsf : this.storeEngine.getStoreFileManager().getStoreFiles()) { 1634 StoreFileReader r = hsf.getReader(); 1635 if (r == null) { 1636 LOG.debug("StoreFile {} has a null Reader", hsf); 1637 continue; 1638 } 1639 this.storeSize.addAndGet(r.length()); 1640 this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes()); 1641 } 1642 } 1643 1644 /* 1645 * @param wantedVersions How many versions were asked for. 1646 * @return wantedVersions or this families' {@link HConstants#VERSIONS}. 1647 */ 1648 int versionsToReturn(final int wantedVersions) { 1649 if (wantedVersions <= 0) { 1650 throw new IllegalArgumentException("Number of versions must be > 0"); 1651 } 1652 // Make sure we do not return more than maximum versions for this store. 1653 int maxVersions = getColumnFamilyDescriptor().getMaxVersions(); 1654 return wantedVersions > maxVersions ? maxVersions : wantedVersions; 1655 } 1656 1657 @Override 1658 public boolean canSplit() { 1659 // Not split-able if we find a reference store file present in the store. 1660 boolean result = !hasReferences(); 1661 if (!result) { 1662 LOG.trace("Not splittable; has references: {}", this); 1663 } 1664 return result; 1665 } 1666 1667 /** 1668 * Determines if Store should be split. 1669 */ 1670 public Optional<byte[]> getSplitPoint() { 1671 this.storeEngine.readLock(); 1672 try { 1673 // Should already be enforced by the split policy! 1674 assert !this.getRegionInfo().isMetaRegion(); 1675 // Not split-able if we find a reference store file present in the store. 1676 if (hasReferences()) { 1677 LOG.trace("Not splittable; has references: {}", this); 1678 return Optional.empty(); 1679 } 1680 return this.storeEngine.getStoreFileManager().getSplitPoint(); 1681 } catch (IOException e) { 1682 LOG.warn("Failed getting store size for {}", this, e); 1683 } finally { 1684 this.storeEngine.readUnlock(); 1685 } 1686 return Optional.empty(); 1687 } 1688 1689 @Override 1690 public long getLastCompactSize() { 1691 return this.lastCompactSize; 1692 } 1693 1694 @Override 1695 public long getSize() { 1696 return storeSize.get(); 1697 } 1698 1699 public void triggerMajorCompaction() { 1700 this.forceMajor = true; 1701 } 1702 1703 ////////////////////////////////////////////////////////////////////////////// 1704 // File administration 1705 ////////////////////////////////////////////////////////////////////////////// 1706 1707 /** 1708 * Return a scanner for both the memstore and the HStore files. Assumes we are not in a 1709 * compaction. 1710 * @param scan Scan to apply when scanning the stores 1711 * @param targetCols columns to scan 1712 * @return a scanner over the current key values 1713 * @throws IOException on failure 1714 */ 1715 public KeyValueScanner getScanner(Scan scan, final NavigableSet<byte[]> targetCols, long readPt) 1716 throws IOException { 1717 storeEngine.readLock(); 1718 try { 1719 ScanInfo scanInfo; 1720 if (this.getCoprocessorHost() != null) { 1721 scanInfo = this.getCoprocessorHost().preStoreScannerOpen(this, scan); 1722 } else { 1723 scanInfo = getScanInfo(); 1724 } 1725 return createScanner(scan, scanInfo, targetCols, readPt); 1726 } finally { 1727 storeEngine.readUnlock(); 1728 } 1729 } 1730 1731 // HMobStore will override this method to return its own implementation. 1732 protected KeyValueScanner createScanner(Scan scan, ScanInfo scanInfo, 1733 NavigableSet<byte[]> targetCols, long readPt) throws IOException { 1734 return scan.isReversed() 1735 ? new ReversedStoreScanner(this, scanInfo, scan, targetCols, readPt) 1736 : new StoreScanner(this, scanInfo, scan, targetCols, readPt); 1737 } 1738 1739 /** 1740 * Recreates the scanners on the current list of active store file scanners 1741 * @param currentFileScanners the current set of active store file scanners 1742 * @param cacheBlocks cache the blocks or not 1743 * @param usePread use pread or not 1744 * @param isCompaction is the scanner for compaction 1745 * @param matcher the scan query matcher 1746 * @param startRow the scan's start row 1747 * @param includeStartRow should the scan include the start row 1748 * @param stopRow the scan's stop row 1749 * @param includeStopRow should the scan include the stop row 1750 * @param readPt the read point of the current scane 1751 * @param includeMemstoreScanner whether the current scanner should include memstorescanner 1752 * @return list of scanners recreated on the current Scanners 1753 */ 1754 public List<KeyValueScanner> recreateScanners(List<KeyValueScanner> currentFileScanners, 1755 boolean cacheBlocks, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher, 1756 byte[] startRow, boolean includeStartRow, byte[] stopRow, boolean includeStopRow, long readPt, 1757 boolean includeMemstoreScanner) throws IOException { 1758 this.storeEngine.readLock(); 1759 try { 1760 Map<String, HStoreFile> name2File = 1761 new HashMap<>(getStorefilesCount() + getCompactedFilesCount()); 1762 for (HStoreFile file : getStorefiles()) { 1763 name2File.put(file.getFileInfo().getActiveFileName(), file); 1764 } 1765 Collection<HStoreFile> compactedFiles = getCompactedFiles(); 1766 for (HStoreFile file : IterableUtils.emptyIfNull(compactedFiles)) { 1767 name2File.put(file.getFileInfo().getActiveFileName(), file); 1768 } 1769 List<HStoreFile> filesToReopen = new ArrayList<>(); 1770 for (KeyValueScanner kvs : currentFileScanners) { 1771 assert kvs.isFileScanner(); 1772 if (kvs.peek() == null) { 1773 continue; 1774 } 1775 filesToReopen.add(name2File.get(kvs.getFilePath().getName())); 1776 } 1777 if (filesToReopen.isEmpty()) { 1778 return null; 1779 } 1780 return getScanners(filesToReopen, cacheBlocks, false, false, matcher, startRow, 1781 includeStartRow, stopRow, includeStopRow, readPt, false, false); 1782 } finally { 1783 this.storeEngine.readUnlock(); 1784 } 1785 } 1786 1787 @Override 1788 public String toString() { 1789 return this.getRegionInfo().getShortNameToLog() + "/" + this.getColumnFamilyName(); 1790 } 1791 1792 @Override 1793 public int getStorefilesCount() { 1794 return this.storeEngine.getStoreFileManager().getStorefileCount(); 1795 } 1796 1797 @Override 1798 public int getCompactedFilesCount() { 1799 return this.storeEngine.getStoreFileManager().getCompactedFilesCount(); 1800 } 1801 1802 private LongStream getStoreFileAgeStream() { 1803 return this.storeEngine.getStoreFileManager().getStoreFiles().stream().filter(sf -> { 1804 if (sf.getReader() == null) { 1805 LOG.debug("StoreFile {} has a null Reader", sf); 1806 return false; 1807 } else { 1808 return true; 1809 } 1810 }).filter(HStoreFile::isHFile).mapToLong(sf -> sf.getFileInfo().getCreatedTimestamp()) 1811 .map(t -> EnvironmentEdgeManager.currentTime() - t); 1812 } 1813 1814 @Override 1815 public OptionalLong getMaxStoreFileAge() { 1816 return getStoreFileAgeStream().max(); 1817 } 1818 1819 @Override 1820 public OptionalLong getMinStoreFileAge() { 1821 return getStoreFileAgeStream().min(); 1822 } 1823 1824 @Override 1825 public OptionalDouble getAvgStoreFileAge() { 1826 return getStoreFileAgeStream().average(); 1827 } 1828 1829 @Override 1830 public long getNumReferenceFiles() { 1831 return this.storeEngine.getStoreFileManager().getStoreFiles().stream() 1832 .filter(HStoreFile::isReference).count(); 1833 } 1834 1835 @Override 1836 public long getNumHFiles() { 1837 return this.storeEngine.getStoreFileManager().getStoreFiles().stream() 1838 .filter(HStoreFile::isHFile).count(); 1839 } 1840 1841 @Override 1842 public long getStoreSizeUncompressed() { 1843 return this.totalUncompressedBytes.get(); 1844 } 1845 1846 @Override 1847 public long getStorefilesSize() { 1848 // Include all StoreFiles 1849 return StoreUtils.getStorefilesSize(this.storeEngine.getStoreFileManager().getStoreFiles(), 1850 sf -> true); 1851 } 1852 1853 @Override 1854 public long getHFilesSize() { 1855 // Include only StoreFiles which are HFiles 1856 return StoreUtils.getStorefilesSize(this.storeEngine.getStoreFileManager().getStoreFiles(), 1857 HStoreFile::isHFile); 1858 } 1859 1860 private long getStorefilesFieldSize(ToLongFunction<StoreFileReader> f) { 1861 return this.storeEngine.getStoreFileManager().getStoreFiles().stream() 1862 .mapToLong(file -> StoreUtils.getStorefileFieldSize(file, f)).sum(); 1863 } 1864 1865 @Override 1866 public long getStorefilesRootLevelIndexSize() { 1867 return getStorefilesFieldSize(StoreFileReader::indexSize); 1868 } 1869 1870 @Override 1871 public long getTotalStaticIndexSize() { 1872 return getStorefilesFieldSize(StoreFileReader::getUncompressedDataIndexSize); 1873 } 1874 1875 @Override 1876 public long getTotalStaticBloomSize() { 1877 return getStorefilesFieldSize(StoreFileReader::getTotalBloomSize); 1878 } 1879 1880 @Override 1881 public MemStoreSize getMemStoreSize() { 1882 return this.memstore.size(); 1883 } 1884 1885 @Override 1886 public int getCompactPriority() { 1887 int priority = this.storeEngine.getStoreFileManager().getStoreCompactionPriority(); 1888 if (priority == PRIORITY_USER) { 1889 LOG.warn("Compaction priority is USER despite there being no user compaction"); 1890 } 1891 return priority; 1892 } 1893 1894 public boolean throttleCompaction(long compactionSize) { 1895 return storeEngine.getCompactionPolicy().throttleCompaction(compactionSize); 1896 } 1897 1898 public HRegion getHRegion() { 1899 return this.region; 1900 } 1901 1902 public RegionCoprocessorHost getCoprocessorHost() { 1903 return this.region.getCoprocessorHost(); 1904 } 1905 1906 @Override 1907 public RegionInfo getRegionInfo() { 1908 return getRegionFileSystem().getRegionInfo(); 1909 } 1910 1911 @Override 1912 public boolean areWritesEnabled() { 1913 return this.region.areWritesEnabled(); 1914 } 1915 1916 @Override 1917 public long getSmallestReadPoint() { 1918 return this.region.getSmallestReadPoint(); 1919 } 1920 1921 /** 1922 * Adds or replaces the specified KeyValues. 1923 * <p> 1924 * For each KeyValue specified, if a cell with the same row, family, and qualifier exists in 1925 * MemStore, it will be replaced. Otherwise, it will just be inserted to MemStore. 1926 * <p> 1927 * This operation is atomic on each KeyValue (row/family/qualifier) but not necessarily atomic 1928 * across all of them. 1929 * @param readpoint readpoint below which we can safely remove duplicate KVs 1930 */ 1931 public void upsert(Iterable<ExtendedCell> cells, long readpoint, MemStoreSizing memstoreSizing) { 1932 this.storeEngine.readLock(); 1933 try { 1934 this.memstore.upsert(cells, readpoint, memstoreSizing); 1935 } finally { 1936 this.storeEngine.readUnlock(); 1937 } 1938 } 1939 1940 public StoreFlushContext createFlushContext(long cacheFlushId, FlushLifeCycleTracker tracker) { 1941 return new StoreFlusherImpl(cacheFlushId, tracker); 1942 } 1943 1944 private final class StoreFlusherImpl implements StoreFlushContext { 1945 1946 private final FlushLifeCycleTracker tracker; 1947 private final StoreFileWriterCreationTracker writerCreationTracker; 1948 private final long cacheFlushSeqNum; 1949 private MemStoreSnapshot snapshot; 1950 private List<Path> tempFiles; 1951 private List<Path> committedFiles; 1952 private long cacheFlushCount; 1953 private long cacheFlushSize; 1954 private long outputFileSize; 1955 1956 private StoreFlusherImpl(long cacheFlushSeqNum, FlushLifeCycleTracker tracker) { 1957 this.cacheFlushSeqNum = cacheFlushSeqNum; 1958 this.tracker = tracker; 1959 this.writerCreationTracker = storeFileWriterCreationTrackerFactory.get(); 1960 } 1961 1962 /** 1963 * This is not thread safe. The caller should have a lock on the region or the store. If 1964 * necessary, the lock can be added with the patch provided in HBASE-10087 1965 */ 1966 @Override 1967 public MemStoreSize prepare() { 1968 // passing the current sequence number of the wal - to allow bookkeeping in the memstore 1969 this.snapshot = memstore.snapshot(); 1970 this.cacheFlushCount = snapshot.getCellsCount(); 1971 this.cacheFlushSize = snapshot.getDataSize(); 1972 committedFiles = new ArrayList<>(1); 1973 return snapshot.getMemStoreSize(); 1974 } 1975 1976 @Override 1977 public void flushCache(MonitoredTask status) throws IOException { 1978 RegionServerServices rsService = region.getRegionServerServices(); 1979 ThroughputController throughputController = 1980 rsService == null ? null : rsService.getFlushThroughputController(); 1981 // it could be null if we do not need to track the creation of store file writer due to 1982 // different SFT implementation. 1983 if (writerCreationTracker != null) { 1984 HStore.this.storeFileWriterCreationTrackers.add(writerCreationTracker); 1985 } 1986 tempFiles = HStore.this.flushCache(cacheFlushSeqNum, snapshot, status, throughputController, 1987 tracker, writerCreationTracker); 1988 } 1989 1990 @Override 1991 public boolean commit(MonitoredTask status) throws IOException { 1992 try { 1993 if (CollectionUtils.isEmpty(this.tempFiles)) { 1994 return false; 1995 } 1996 status.setStatus("Flushing " + this + ": reopening flushed file"); 1997 List<HStoreFile> storeFiles = storeEngine.commitStoreFiles(tempFiles, false, false); 1998 for (HStoreFile sf : storeFiles) { 1999 StoreFileReader r = sf.getReader(); 2000 if (LOG.isInfoEnabled()) { 2001 LOG.info("Added {}, entries={}, sequenceid={}, filesize={}", sf, r.getEntries(), 2002 cacheFlushSeqNum, TraditionalBinaryPrefix.long2String(r.length(), "", 1)); 2003 } 2004 outputFileSize += r.length(); 2005 storeSize.addAndGet(r.length()); 2006 totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes()); 2007 committedFiles.add(sf.getPath()); 2008 } 2009 2010 flushedCellsCount.addAndGet(cacheFlushCount); 2011 flushedCellsSize.addAndGet(cacheFlushSize); 2012 flushedOutputFileSize.addAndGet(outputFileSize); 2013 // call coprocessor after we have done all the accounting above 2014 for (HStoreFile sf : storeFiles) { 2015 if (getCoprocessorHost() != null) { 2016 getCoprocessorHost().postFlush(HStore.this, sf, tracker); 2017 } 2018 } 2019 // Add new file to store files. Clear snapshot too while we have the Store write lock. 2020 return completeFlush(storeFiles, snapshot.getId()); 2021 } finally { 2022 if (writerCreationTracker != null) { 2023 HStore.this.storeFileWriterCreationTrackers.remove(writerCreationTracker); 2024 } 2025 } 2026 } 2027 2028 @Override 2029 public long getOutputFileSize() { 2030 return outputFileSize; 2031 } 2032 2033 @Override 2034 public List<Path> getCommittedFiles() { 2035 return committedFiles; 2036 } 2037 2038 /** 2039 * Similar to commit, but called in secondary region replicas for replaying the flush cache from 2040 * primary region. Adds the new files to the store, and drops the snapshot depending on 2041 * dropMemstoreSnapshot argument. 2042 * @param fileNames names of the flushed files 2043 * @param dropMemstoreSnapshot whether to drop the prepared memstore snapshot 2044 */ 2045 @Override 2046 public void replayFlush(List<String> fileNames, boolean dropMemstoreSnapshot) 2047 throws IOException { 2048 List<HStoreFile> storeFiles = new ArrayList<>(fileNames.size()); 2049 for (String file : fileNames) { 2050 // open the file as a store file (hfile link, etc) 2051 StoreFileTracker sft = StoreFileTrackerFactory.create(conf, false, storeContext); 2052 StoreFileInfo storeFileInfo = 2053 getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), file, sft); 2054 HStoreFile storeFile = storeEngine.createStoreFileAndReader(storeFileInfo); 2055 storeFiles.add(storeFile); 2056 HStore.this.storeSize.addAndGet(storeFile.getReader().length()); 2057 HStore.this.totalUncompressedBytes 2058 .addAndGet(storeFile.getReader().getTotalUncompressedBytes()); 2059 if (LOG.isInfoEnabled()) { 2060 LOG.info(this + " added " + storeFile + ", entries=" + storeFile.getReader().getEntries() 2061 + ", sequenceid=" + storeFile.getReader().getSequenceID() + ", filesize=" 2062 + TraditionalBinaryPrefix.long2String(storeFile.getReader().length(), "", 1)); 2063 } 2064 } 2065 2066 long snapshotId = -1; // -1 means do not drop 2067 if (dropMemstoreSnapshot && snapshot != null) { 2068 snapshotId = snapshot.getId(); 2069 } 2070 HStore.this.completeFlush(storeFiles, snapshotId); 2071 } 2072 2073 /** 2074 * Abort the snapshot preparation. Drops the snapshot if any. 2075 */ 2076 @Override 2077 public void abort() throws IOException { 2078 if (snapshot != null) { 2079 HStore.this.completeFlush(Collections.emptyList(), snapshot.getId()); 2080 } 2081 } 2082 } 2083 2084 @Override 2085 public boolean needsCompaction() { 2086 List<HStoreFile> filesCompactingClone = null; 2087 synchronized (filesCompacting) { 2088 filesCompactingClone = Lists.newArrayList(filesCompacting); 2089 } 2090 return this.storeEngine.needsCompaction(filesCompactingClone); 2091 } 2092 2093 /** 2094 * Used for tests. 2095 * @return cache configuration for this Store. 2096 */ 2097 public CacheConfig getCacheConfig() { 2098 return storeContext.getCacheConf(); 2099 } 2100 2101 public static final long FIXED_OVERHEAD = ClassSize.estimateBase(HStore.class, false); 2102 2103 public static final long DEEP_OVERHEAD = ClassSize.align( 2104 FIXED_OVERHEAD + ClassSize.OBJECT + ClassSize.REENTRANT_LOCK + ClassSize.CONCURRENT_SKIPLISTMAP 2105 + ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY + ClassSize.OBJECT + ScanInfo.FIXED_OVERHEAD); 2106 2107 @Override 2108 public long heapSize() { 2109 MemStoreSize memstoreSize = this.memstore.size(); 2110 return DEEP_OVERHEAD + memstoreSize.getHeapSize() + storeContext.heapSize(); 2111 } 2112 2113 @Override 2114 public CellComparator getComparator() { 2115 return storeContext.getComparator(); 2116 } 2117 2118 public ScanInfo getScanInfo() { 2119 return scanInfo; 2120 } 2121 2122 /** 2123 * Set scan info, used by test 2124 * @param scanInfo new scan info to use for test 2125 */ 2126 void setScanInfo(ScanInfo scanInfo) { 2127 this.scanInfo = scanInfo; 2128 } 2129 2130 @Override 2131 public boolean hasTooManyStoreFiles() { 2132 return getStorefilesCount() > this.blockingFileCount; 2133 } 2134 2135 @Override 2136 public long getFlushedCellsCount() { 2137 return flushedCellsCount.get(); 2138 } 2139 2140 @Override 2141 public long getFlushedCellsSize() { 2142 return flushedCellsSize.get(); 2143 } 2144 2145 @Override 2146 public long getFlushedOutputFileSize() { 2147 return flushedOutputFileSize.get(); 2148 } 2149 2150 @Override 2151 public long getCompactedCellsCount() { 2152 return compactedCellsCount.get(); 2153 } 2154 2155 @Override 2156 public long getCompactedCellsSize() { 2157 return compactedCellsSize.get(); 2158 } 2159 2160 @Override 2161 public long getMajorCompactedCellsCount() { 2162 return majorCompactedCellsCount.get(); 2163 } 2164 2165 @Override 2166 public long getMajorCompactedCellsSize() { 2167 return majorCompactedCellsSize.get(); 2168 } 2169 2170 public void updateCompactedMetrics(boolean isMajor, CompactionProgress progress) { 2171 if (isMajor) { 2172 majorCompactedCellsCount.addAndGet(progress.getTotalCompactingKVs()); 2173 majorCompactedCellsSize.addAndGet(progress.totalCompactedSize); 2174 } else { 2175 compactedCellsCount.addAndGet(progress.getTotalCompactingKVs()); 2176 compactedCellsSize.addAndGet(progress.totalCompactedSize); 2177 } 2178 } 2179 2180 /** 2181 * Returns the StoreEngine that is backing this concrete implementation of Store. 2182 * @return Returns the {@link StoreEngine} object used internally inside this HStore object. 2183 */ 2184 public StoreEngine<?, ?, ?, ?> getStoreEngine() { 2185 return this.storeEngine; 2186 } 2187 2188 protected OffPeakHours getOffPeakHours() { 2189 return this.offPeakHours; 2190 } 2191 2192 @Override 2193 public void onConfigurationChange(Configuration conf) { 2194 Configuration storeConf = StoreUtils.createStoreConfiguration(conf, region.getTableDescriptor(), 2195 getColumnFamilyDescriptor()); 2196 this.conf = storeConf; 2197 this.storeEngine.compactionPolicy.setConf(storeConf); 2198 this.offPeakHours = OffPeakHours.getInstance(storeConf); 2199 } 2200 2201 /** 2202 * {@inheritDoc} 2203 */ 2204 @Override 2205 public void registerChildren(ConfigurationManager manager) { 2206 CacheConfig cacheConfig = this.storeContext.getCacheConf(); 2207 if (cacheConfig != null) { 2208 manager.registerObserver(cacheConfig); 2209 } 2210 } 2211 2212 /** 2213 * {@inheritDoc} 2214 */ 2215 @Override 2216 public void deregisterChildren(ConfigurationManager manager) { 2217 // No children to deregister 2218 } 2219 2220 @Override 2221 public double getCompactionPressure() { 2222 return storeEngine.getStoreFileManager().getCompactionPressure(); 2223 } 2224 2225 @Override 2226 public boolean isPrimaryReplicaStore() { 2227 return getRegionInfo().getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID; 2228 } 2229 2230 /** 2231 * Sets the store up for a region level snapshot operation. 2232 * @see #postSnapshotOperation() 2233 */ 2234 public void preSnapshotOperation() { 2235 archiveLock.lock(); 2236 } 2237 2238 /** 2239 * Perform tasks needed after the completion of snapshot operation. 2240 * @see #preSnapshotOperation() 2241 */ 2242 public void postSnapshotOperation() { 2243 archiveLock.unlock(); 2244 } 2245 2246 /** 2247 * Closes and archives the compacted files under this store 2248 */ 2249 public synchronized void closeAndArchiveCompactedFiles() throws IOException { 2250 // ensure other threads do not attempt to archive the same files on close() 2251 archiveLock.lock(); 2252 try { 2253 storeEngine.readLock(); 2254 Collection<HStoreFile> copyCompactedfiles = null; 2255 try { 2256 Collection<HStoreFile> compactedfiles = 2257 this.getStoreEngine().getStoreFileManager().getCompactedfiles(); 2258 if (CollectionUtils.isNotEmpty(compactedfiles)) { 2259 // Do a copy under read lock 2260 copyCompactedfiles = new ArrayList<>(compactedfiles); 2261 } else { 2262 LOG.trace("No compacted files to archive"); 2263 } 2264 } finally { 2265 storeEngine.readUnlock(); 2266 } 2267 if (CollectionUtils.isNotEmpty(copyCompactedfiles)) { 2268 removeCompactedfiles(copyCompactedfiles, true); 2269 } 2270 } finally { 2271 archiveLock.unlock(); 2272 } 2273 } 2274 2275 /** 2276 * Archives and removes the compacted files 2277 * @param compactedfiles The compacted files in this store that are not active in reads 2278 * @param evictOnClose true if blocks should be evicted from the cache when an HFile reader is 2279 * closed, false if not 2280 */ 2281 private void removeCompactedfiles(Collection<HStoreFile> compactedfiles, boolean evictOnClose) 2282 throws IOException { 2283 final List<HStoreFile> filesToRemove = new ArrayList<>(compactedfiles.size()); 2284 final List<Long> storeFileSizes = new ArrayList<>(compactedfiles.size()); 2285 for (final HStoreFile file : compactedfiles) { 2286 synchronized (file) { 2287 try { 2288 StoreFileReader r = file.getReader(); 2289 if (r == null) { 2290 LOG.debug("The file {} was closed but still not archived", file); 2291 // HACK: Temporarily re-open the reader so we can get the size of the file. Ideally, 2292 // we should know the size of an HStoreFile without having to ask the HStoreFileReader 2293 // for that. 2294 long length = getStoreFileSize(file); 2295 filesToRemove.add(file); 2296 storeFileSizes.add(length); 2297 continue; 2298 } 2299 2300 if (file.isCompactedAway() && !file.isReferencedInReads()) { 2301 // Even if deleting fails we need not bother as any new scanners won't be 2302 // able to use the compacted file as the status is already compactedAway 2303 LOG.trace("Closing and archiving the file {}", file); 2304 // Copy the file size before closing the reader 2305 final long length = r.length(); 2306 r.close(evictOnClose); 2307 // Just close and return 2308 filesToRemove.add(file); 2309 // Only add the length if we successfully added the file to `filesToRemove` 2310 storeFileSizes.add(length); 2311 } else { 2312 LOG.info("Can't archive compacted file " + file.getPath() 2313 + " because of either isCompactedAway=" + file.isCompactedAway() 2314 + " or file has reference, isReferencedInReads=" + file.isReferencedInReads() 2315 + ", refCount=" + r.getRefCount() + ", skipping for now."); 2316 } 2317 } catch (Exception e) { 2318 LOG.error("Exception while trying to close the compacted store file {}", file.getPath(), 2319 e); 2320 } 2321 } 2322 } 2323 if (this.isPrimaryReplicaStore()) { 2324 // Only the primary region is allowed to move the file to archive. 2325 // The secondary region does not move the files to archive. Any active reads from 2326 // the secondary region will still work because the file as such has active readers on it. 2327 if (!filesToRemove.isEmpty()) { 2328 LOG.debug("Moving the files {} to archive", filesToRemove); 2329 // Only if this is successful it has to be removed 2330 try { 2331 getRegionFileSystem().removeStoreFiles(this.getColumnFamilyDescriptor().getNameAsString(), 2332 filesToRemove); 2333 } catch (FailedArchiveException fae) { 2334 // Even if archiving some files failed, we still need to clear out any of the 2335 // files which were successfully archived. Otherwise we will receive a 2336 // FileNotFoundException when we attempt to re-archive them in the next go around. 2337 Collection<Path> failedFiles = fae.getFailedFiles(); 2338 Iterator<HStoreFile> iter = filesToRemove.iterator(); 2339 Iterator<Long> sizeIter = storeFileSizes.iterator(); 2340 while (iter.hasNext()) { 2341 sizeIter.next(); 2342 if (failedFiles.contains(iter.next().getPath())) { 2343 iter.remove(); 2344 sizeIter.remove(); 2345 } 2346 } 2347 if (!filesToRemove.isEmpty()) { 2348 clearCompactedfiles(filesToRemove); 2349 } 2350 throw fae; 2351 } 2352 } 2353 } 2354 if (!filesToRemove.isEmpty()) { 2355 // Clear the compactedfiles from the store file manager 2356 clearCompactedfiles(filesToRemove); 2357 // Try to send report of this archival to the Master for updating quota usage faster 2358 reportArchivedFilesForQuota(filesToRemove, storeFileSizes); 2359 } 2360 } 2361 2362 /** 2363 * Computes the length of a store file without succumbing to any errors along the way. If an error 2364 * is encountered, the implementation returns {@code 0} instead of the actual size. 2365 * @param file The file to compute the size of. 2366 * @return The size in bytes of the provided {@code file}. 2367 */ 2368 long getStoreFileSize(HStoreFile file) { 2369 long length = 0; 2370 try { 2371 file.initReader(); 2372 length = file.getReader().length(); 2373 } catch (IOException e) { 2374 LOG.trace("Failed to open reader when trying to compute store file size for {}, ignoring", 2375 file, e); 2376 } finally { 2377 try { 2378 file.closeStoreFile( 2379 file.getCacheConf() != null ? file.getCacheConf().shouldEvictOnClose() : true); 2380 } catch (IOException e) { 2381 LOG.trace("Failed to close reader after computing store file size for {}, ignoring", file, 2382 e); 2383 } 2384 } 2385 return length; 2386 } 2387 2388 public Long preFlushSeqIDEstimation() { 2389 return memstore.preFlushSeqIDEstimation(); 2390 } 2391 2392 @Override 2393 public boolean isSloppyMemStore() { 2394 return this.memstore.isSloppy(); 2395 } 2396 2397 private void clearCompactedfiles(List<HStoreFile> filesToRemove) throws IOException { 2398 LOG.trace("Clearing the compacted file {} from this store", filesToRemove); 2399 storeEngine.removeCompactedFiles(filesToRemove); 2400 } 2401 2402 void reportArchivedFilesForQuota(List<? extends StoreFile> archivedFiles, List<Long> fileSizes) { 2403 // Sanity check from the caller 2404 if (archivedFiles.size() != fileSizes.size()) { 2405 throw new RuntimeException("Coding error: should never see lists of varying size"); 2406 } 2407 RegionServerServices rss = this.region.getRegionServerServices(); 2408 if (rss == null) { 2409 return; 2410 } 2411 List<Entry<String, Long>> filesWithSizes = new ArrayList<>(archivedFiles.size()); 2412 Iterator<Long> fileSizeIter = fileSizes.iterator(); 2413 for (StoreFile storeFile : archivedFiles) { 2414 final long fileSize = fileSizeIter.next(); 2415 if (storeFile.isHFile() && fileSize != 0) { 2416 filesWithSizes.add(Maps.immutableEntry(storeFile.getPath().getName(), fileSize)); 2417 } 2418 } 2419 if (LOG.isTraceEnabled()) { 2420 LOG.trace("Files archived: " + archivedFiles + ", reporting the following to the Master: " 2421 + filesWithSizes); 2422 } 2423 boolean success = rss.reportFileArchivalForQuotas(getTableName(), filesWithSizes); 2424 if (!success) { 2425 LOG.warn("Failed to report archival of files: " + filesWithSizes); 2426 } 2427 } 2428 2429 @Override 2430 public int getCurrentParallelPutCount() { 2431 return currentParallelPutCount.get(); 2432 } 2433 2434 public int getStoreRefCount() { 2435 return this.storeEngine.getStoreFileManager().getStoreFiles().stream() 2436 .filter(sf -> sf.getReader() != null).filter(HStoreFile::isHFile) 2437 .mapToInt(HStoreFile::getRefCount).sum(); 2438 } 2439 2440 /** Returns get maximum ref count of storeFile among all compacted HStore Files for the HStore */ 2441 public int getMaxCompactedStoreFileRefCount() { 2442 OptionalInt maxCompactedStoreFileRefCount = this.storeEngine.getStoreFileManager() 2443 .getCompactedfiles().stream().filter(sf -> sf.getReader() != null).filter(HStoreFile::isHFile) 2444 .mapToInt(HStoreFile::getRefCount).max(); 2445 return maxCompactedStoreFileRefCount.isPresent() ? maxCompactedStoreFileRefCount.getAsInt() : 0; 2446 } 2447 2448 @Override 2449 public long getMemstoreOnlyRowReadsCount() { 2450 return memstoreOnlyRowReadsCount.sum(); 2451 } 2452 2453 @Override 2454 public long getMixedRowReadsCount() { 2455 return mixedRowReadsCount.sum(); 2456 } 2457 2458 @Override 2459 public Configuration getReadOnlyConfiguration() { 2460 return new ReadOnlyConfiguration(this.conf); 2461 } 2462 2463 void updateMetricsStore(boolean memstoreRead) { 2464 if (memstoreRead) { 2465 memstoreOnlyRowReadsCount.increment(); 2466 } else { 2467 mixedRowReadsCount.increment(); 2468 } 2469 } 2470 2471 /** 2472 * Return the storefiles which are currently being written to. Mainly used by 2473 * {@link BrokenStoreFileCleaner} to prevent deleting the these files as they are not present in 2474 * SFT yet. 2475 */ 2476 public Set<Path> getStoreFilesBeingWritten() { 2477 return storeFileWriterCreationTrackers.stream().flatMap(t -> t.get().stream()) 2478 .collect(Collectors.toSet()); 2479 } 2480 2481 @Override 2482 public long getBloomFilterRequestsCount() { 2483 return storeEngine.getBloomFilterMetrics().getRequestsCount(); 2484 } 2485 2486 @Override 2487 public long getBloomFilterNegativeResultsCount() { 2488 return storeEngine.getBloomFilterMetrics().getNegativeResultsCount(); 2489 } 2490 2491 @Override 2492 public long getBloomFilterEligibleRequestsCount() { 2493 return storeEngine.getBloomFilterMetrics().getEligibleRequestsCount(); 2494 } 2495}