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.util.compaction;
019
020import java.io.IOException;
021import java.util.Collection;
022import java.util.List;
023import java.util.Optional;
024import java.util.Set;
025import org.apache.hadoop.fs.FileStatus;
026import org.apache.hadoop.fs.FileSystem;
027import org.apache.hadoop.fs.Path;
028import org.apache.hadoop.hbase.client.Admin;
029import org.apache.hadoop.hbase.client.Connection;
030import org.apache.hadoop.hbase.client.RegionInfo;
031import org.apache.hadoop.hbase.client.TableDescriptor;
032import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
033import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
034import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
035import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
036import org.apache.hadoop.hbase.util.CommonFSUtils;
037import org.apache.hadoop.hbase.util.FSUtils;
038import org.apache.yetus.audience.InterfaceAudience;
039import org.slf4j.Logger;
040import org.slf4j.LoggerFactory;
041
042import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
043
044@InterfaceAudience.Private
045class MajorCompactionRequest {
046
047  private static final Logger LOG = LoggerFactory.getLogger(MajorCompactionRequest.class);
048
049  protected final Connection connection;
050  protected final RegionInfo region;
051  private Set<String> stores;
052
053  MajorCompactionRequest(Connection connection, RegionInfo region) {
054    this.connection = connection;
055    this.region = region;
056  }
057
058  MajorCompactionRequest(Connection connection, RegionInfo region, Set<String> stores) {
059    this(connection, region);
060    this.stores = stores;
061  }
062
063  static Optional<MajorCompactionRequest> newRequest(Connection connection, RegionInfo info,
064    Set<String> stores, long timestamp) throws IOException {
065    MajorCompactionRequest request = new MajorCompactionRequest(connection, info, stores);
066    return request.createRequest(connection, stores, timestamp);
067  }
068
069  RegionInfo getRegion() {
070    return region;
071  }
072
073  Set<String> getStores() {
074    return stores;
075  }
076
077  void setStores(Set<String> stores) {
078    this.stores = stores;
079  }
080
081  Optional<MajorCompactionRequest> createRequest(Connection connection, Set<String> stores,
082    long timestamp) throws IOException {
083    Set<String> familiesToCompact = getStoresRequiringCompaction(stores, timestamp);
084    MajorCompactionRequest request = null;
085    if (!familiesToCompact.isEmpty()) {
086      request = new MajorCompactionRequest(connection, region, familiesToCompact);
087    }
088    return Optional.ofNullable(request);
089  }
090
091  Set<String> getStoresRequiringCompaction(Set<String> requestedStores, long timestamp)
092    throws IOException {
093    HRegionFileSystem fileSystem = getFileSystem();
094    Set<String> familiesToCompact = Sets.newHashSet();
095    for (String family : requestedStores) {
096      if (shouldCFBeCompacted(fileSystem, family, timestamp)) {
097        familiesToCompact.add(family);
098      }
099    }
100    return familiesToCompact;
101  }
102
103  boolean shouldCFBeCompacted(HRegionFileSystem fileSystem, String family, long ts)
104    throws IOException {
105    // do we have any store files?
106    StoreFileTracker sft = getStoreFileTracker(family, fileSystem);
107    List<StoreFileInfo> storeFiles = sft.load();
108    if (storeFiles == null) {
109      LOG.info("Excluding store: " + family + " for compaction for region:  "
110        + fileSystem.getRegionInfo().getEncodedName(), " has no store files");
111      return false;
112    }
113    // check for reference files
114    if (sft.hasReferences() && familyHasReferenceFile(fileSystem, family, ts)) {
115      LOG.info("Including store: " + family + " with: " + storeFiles.size()
116        + " files for compaction for region: " + fileSystem.getRegionInfo().getEncodedName());
117      return true;
118    }
119    // check store file timestamps
120    boolean includeStore = this.shouldIncludeStore(fileSystem, family, storeFiles, ts);
121    if (!includeStore) {
122      LOG.info("Excluding store: " + family + " for compaction for region:  "
123        + fileSystem.getRegionInfo().getEncodedName() + " already compacted");
124    }
125    return includeStore;
126  }
127
128  public StoreFileTracker getStoreFileTracker(String family, HRegionFileSystem fileSystem)
129    throws IOException {
130    TableDescriptor htd = connection.getTable(getRegion().getTable()).getDescriptor();
131    return StoreFileTrackerFactory.create(connection.getConfiguration(), htd,
132      htd.getColumnFamily(family.getBytes()), fileSystem, false);
133  }
134
135  protected boolean shouldIncludeStore(HRegionFileSystem fileSystem, String family,
136    Collection<StoreFileInfo> storeFiles, long ts) throws IOException {
137
138    for (StoreFileInfo storeFile : storeFiles) {
139      if (storeFile.getModificationTime() < ts) {
140        LOG.info("Including store: " + family + " with: " + storeFiles.size()
141          + " files for compaction for region: " + fileSystem.getRegionInfo().getEncodedName());
142        return true;
143      }
144    }
145    return false;
146  }
147
148  protected boolean familyHasReferenceFile(HRegionFileSystem fileSystem, String family, long ts)
149    throws IOException {
150    List<Path> referenceFiles =
151      getReferenceFilePaths(fileSystem.getFileSystem(), fileSystem.getStoreDir(family));
152    for (Path referenceFile : referenceFiles) {
153      FileStatus status = fileSystem.getFileSystem().getFileLinkStatus(referenceFile);
154      if (status.getModificationTime() < ts) {
155        LOG.info("Including store: " + family + " for compaction for region:  "
156          + fileSystem.getRegionInfo().getEncodedName() + " (reference store files)");
157        return true;
158      }
159    }
160    return false;
161
162  }
163
164  List<Path> getReferenceFilePaths(FileSystem fileSystem, Path familyDir) throws IOException {
165    return FSUtils.getReferenceFilePaths(fileSystem, familyDir);
166  }
167
168  HRegionFileSystem getFileSystem() throws IOException {
169    try (Admin admin = connection.getAdmin()) {
170      return HRegionFileSystem.openRegionFromFileSystem(admin.getConfiguration(),
171        CommonFSUtils.getCurrentFileSystem(admin.getConfiguration()), CommonFSUtils.getTableDir(
172          CommonFSUtils.getRootDir(admin.getConfiguration()), region.getTable()),
173        region, true);
174    }
175  }
176
177  @Override
178  public String toString() {
179    return "region: " + region.getEncodedName() + " store(s): " + stores;
180  }
181}