@Override
public void preFlushScannerOpen(
    org.apache.hadoop.hbase.coprocessor.ObserverContext<RegionCoprocessorEnvironment> c,
    Store store, org.apache.hadoop.hbase.regionserver.ScanOptions options,
    org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker tracker) throws IOException {
  if (cache.getLatestState() != null) {
    options.readAllVersions();
  }
}
 

@Override
public void preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanType scanType, ScanOptions options, CompactionLifeCycleTracker tracker,
    CompactionRequest request) throws IOException {
  if (cache.getLatestState() != null) {
    options.readAllVersions();
  }
}
 

@Override
public void preStoreScannerOpen(ObserverContext<RegionCoprocessorEnvironment> ctx,
  Store store, ScanOptions options) throws IOException {
  if (options.getScan().getTimeRange().isAllTime()) {
    setScanOptions(options);
  }
}
 

private void resetTTL(ScanOptions options) {
  OptionalLong expireBefore = getExpireBefore();
  if (!expireBefore.isPresent()) {
    return;
  }
  options.setTTL(EnvironmentEdgeManager.currentTime() - expireBefore.getAsLong());
}
 

@Override
public void preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
  ScanType scanType, ScanOptions options, CompactionLifeCycleTracker tracker,
  CompactionRequest request) throws IOException {
  setScanOptions(options);
}
 

public void preFlushScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
  ScanOptions options,FlushLifeCycleTracker tracker) throws IOException {
  setScanOptions(options);
}
 

public void preMemStoreCompactionCompactScannerOpen(
  ObserverContext<RegionCoprocessorEnvironment> c, Store store, ScanOptions options)
  throws IOException {
  setScanOptions(options);
}
 

private void setScanOptions(ScanOptions options) {
  options.setMaxVersions(TestRegionCoprocessorHost.MAX_VERSIONS);
  options.setMinVersions(TestRegionCoprocessorHost.MIN_VERSIONS);
  options.setKeepDeletedCells(KeepDeletedCells.TRUE);
  options.setTTL(TestRegionCoprocessorHost.TTL);
}
 

@Override
public void preFlushScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanOptions options, FlushLifeCycleTracker tracker) throws IOException {
  options.readAllVersions();
}
 

@Override
public void preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanType scanType, ScanOptions options, CompactionLifeCycleTracker tracker,
    CompactionRequest request) throws IOException {
  options.readAllVersions();
}
 

@Override
public void preMemStoreCompactionCompactScannerOpen(
    ObserverContext<RegionCoprocessorEnvironment> c, Store store, ScanOptions options)
    throws IOException {
  options.readAllVersions();
}
 

@Override
public void preStoreScannerOpen(ObserverContext<RegionCoprocessorEnvironment> ctx, Store store,
    ScanOptions options) throws IOException {
  options.readAllVersions();
}
 

@Override
public void preFlushScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanOptions options, FlushLifeCycleTracker tracker) throws IOException {
  resetTTL(options);
}
 

@Override
public void preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanType scanType, ScanOptions options, CompactionLifeCycleTracker tracker,
    CompactionRequest request) throws IOException {
  resetTTL(options);
}
 

@Override
public void preCompactScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Store store,
        ScanType scanType, ScanOptions options, CompactionLifeCycleTracker tracker,
        final CompactionRequest request) throws IOException {
    // Compaction and split upcalls run with the effective user context of the requesting user.
    // This will lead to failure of cross cluster RPC if the effective user is not
    // the login user. Switch to the login user context to ensure we have the expected
    // security context.
    final String fullTableName = c.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
    // since we will make a call to syscat, do nothing if we are compacting syscat itself
    if (request.isMajor() && !PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME.equals(fullTableName)) {
        User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
            @Override
            public Void run() throws Exception {
                // If the index is disabled, keep the deleted cells so the rebuild doesn't corrupt the index
                try (PhoenixConnection conn =
                        QueryUtil.getConnectionOnServer(compactionConfig).unwrap(PhoenixConnection.class)) {
                    PTable table = PhoenixRuntime.getTableNoCache(conn, fullTableName);
                    List<PTable> indexes = PTableType.INDEX.equals(table.getType()) ? Lists.newArrayList(table) : table.getIndexes();
                    // FIXME need to handle views and indexes on views as well
                    for (PTable index : indexes) {
                        if (index.getIndexDisableTimestamp() != 0) {
                            LOGGER.info(
                                "Modifying major compaction scanner to retain deleted cells for a table with disabled index: "
                                        + fullTableName);
                            options.setKeepDeletedCells(KeepDeletedCells.TRUE);
                            options.readAllVersions();
                            options.setTTL(Long.MAX_VALUE);
                        }
                    }
                } catch (Exception e) {
                    if (e instanceof TableNotFoundException) {
                        LOGGER.debug("Ignoring HBase table that is not a Phoenix table: " + fullTableName);
                        // non-Phoenix HBase tables won't be found, do nothing
                    } else {
                        LOGGER.error("Unable to modify compaction scanner to retain deleted cells for a table with disabled Index; "
                                + fullTableName,
                                e);
                    }
                }
                return null;
            }
        });
    }
}
 

/**
 * Called before we open store scanner for flush. You can use the {@code options} to change max
 * versions and TTL for the scanner being opened.
 * @param c the environment provided by the region server
 * @param store the store where flush is being requested
 * @param options used to change max versions and TTL for the scanner being opened
 */
default void preFlushScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanOptions options,FlushLifeCycleTracker tracker) throws IOException {}
 

/**
 * Called before we open store scanner for in memory compaction. You can use the {@code options}
 * to change max versions and TTL for the scanner being opened. Notice that this method will only
 * be called when you use {@code eager} mode. For {@code basic} mode we will not drop any cells
 * thus we do not open a store scanner.
 * @param c the environment provided by the region server
 * @param store the store where in memory compaction is being requested
 * @param options used to change max versions and TTL for the scanner being opened
 */
default void preMemStoreCompactionCompactScannerOpen(
    ObserverContext<RegionCoprocessorEnvironment> c, Store store, ScanOptions options)
    throws IOException {}
 

/**
 * Called before we open store scanner for compaction. You can use the {@code options} to change max
 * versions and TTL for the scanner being opened.
 * @param c the environment provided by the region server
 * @param store the store being compacted
 * @param scanType type of Scan
 * @param options used to change max versions and TTL for the scanner being opened
 * @param tracker tracker used to track the life cycle of a compaction
 * @param request the requested compaction
 */
default void preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    ScanType scanType, ScanOptions options, CompactionLifeCycleTracker tracker,
    CompactionRequest request) throws IOException {}
 

/**
 * Called before a store opens a new scanner.
 * <p>
 * This hook is called when a "user" scanner is opened. Use {@code preFlushScannerOpen} and
 * {@code preCompactScannerOpen} to inject flush/compaction.
 * <p>
 * Notice that, this method is used to change the inherent max versions and TTL for a Store. For
 * example, you can change the max versions option for a {@link Scan} object to 10 in
 * {@code preScannerOpen}, but if the max versions config on the Store is 1, then you still can
 * only read 1 version. You need also to inject here to change the max versions to 10 if you want
 * to get more versions.
 * @param ctx the environment provided by the region server
 * @param store the store which we want to get scanner from
 * @param options used to change max versions and TTL for the scanner being opened
 * @see #preFlushScannerOpen(ObserverContext, Store, ScanOptions, FlushLifeCycleTracker)
 * @see #preCompactScannerOpen(ObserverContext, Store, ScanType, ScanOptions,
 *      CompactionLifeCycleTracker, CompactionRequest)
 */
default void preStoreScannerOpen(ObserverContext<RegionCoprocessorEnvironment> ctx, Store store,
    ScanOptions options) throws IOException {}