hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anoopsamj...@apache.org
Subject hbase git commit: HBASE-16962: Add readPoint to preCompactScannerOpen() and preFlushScannerOpen() API
Date Fri, 11 Nov 2016 05:40:19 GMT
Repository: hbase
Updated Branches:
  refs/heads/branch-1 18b31fdd3 -> 44ab659b9


HBASE-16962: Add readPoint to preCompactScannerOpen() and preFlushScannerOpen() API

Signed-off-by: anoopsamjohn <anoopsamjohn@gmail.com>


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/44ab659b
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/44ab659b
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/44ab659b

Branch: refs/heads/branch-1
Commit: 44ab659b933afed2df323b031181fbcb52c85b61
Parents: 18b31fd
Author: thiruvel <thiruvel@yahoo-inc.com>
Authored: Wed Nov 9 11:02:41 2016 -0800
Committer: anoopsamjohn <anoopsamjohn@gmail.com>
Committed: Fri Nov 11 11:09:55 2016 +0530

----------------------------------------------------------------------
 .../hbase/coprocessor/BaseRegionObserver.java   | 14 +++++
 .../hbase/coprocessor/RegionObserver.java       | 60 ++++++++++++++++++--
 .../regionserver/RegionCoprocessorHost.java     | 13 +++--
 .../hadoop/hbase/regionserver/StoreFlusher.java |  3 +-
 .../regionserver/compactions/Compactor.java     | 10 ++--
 5 files changed, 85 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/44ab659b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
index eb2fc28..a0a91bc 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
@@ -103,6 +103,13 @@ public class BaseRegionObserver implements RegionObserver {
   }
 
   @Override
+  public InternalScanner preFlushScannerOpen(final ObserverContext<RegionCoprocessorEnvironment>
c,
+      final Store store, final KeyValueScanner memstoreScanner, final InternalScanner s,
+      final long readPoint) throws IOException {
+    return preFlushScannerOpen(c, store, memstoreScanner, s);
+  }
+
+  @Override
   public void preFlush(ObserverContext<RegionCoprocessorEnvironment> e) throws IOException
{
   }
 
@@ -212,6 +219,13 @@ public class BaseRegionObserver implements RegionObserver {
   }
 
   @Override
+  public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment>
c,
+      Store store, List<? extends KeyValueScanner> scanners, ScanType scanType, long
earliestPutTs,
+      InternalScanner s, CompactionRequest request, long readPoint) throws IOException {
+    return preCompactScannerOpen(c, store, scanners, scanType, earliestPutTs, s, request);
+  }
+
+  @Override
   public void postCompact(ObserverContext<RegionCoprocessorEnvironment> e, final Store
store,
       final StoreFile resultFile) throws IOException {
   }

http://git-wip-us.apache.org/repos/asf/hbase/blob/44ab659b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
index 42d5cdb..7cd1be5 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
@@ -112,12 +112,35 @@ public interface RegionObserver extends Coprocessor {
    * @return the scanner to use during the flush.  {@code null} if the default implementation
    * is to be used.
    * @throws IOException if an error occurred on the coprocessor
+   * @deprecated Use {@link #preFlushScannerOpen(ObserverContext, Store, KeyValueScanner,
+   *             InternalScanner, long)}
    */
+  @Deprecated
   InternalScanner preFlushScannerOpen(final ObserverContext<RegionCoprocessorEnvironment>
c,
       final Store store, final KeyValueScanner memstoreScanner, final InternalScanner s)
       throws IOException;
 
   /**
+   * Called before a memstore is flushed to disk and prior to creating the scanner to read
from
+   * the memstore.  To override or modify how a memstore is flushed,
+   * implementing classes can return a new scanner to provide the KeyValues to be
+   * stored into the new {@code StoreFile} or null to perform the default processing.
+   * Calling {@link org.apache.hadoop.hbase.coprocessor.ObserverContext#bypass()} has no
+   * effect in this hook.
+   * @param c the environment provided by the region server
+   * @param store the store being flushed
+   * @param memstoreScanner the scanner for the memstore that is flushed
+   * @param s the base scanner, if not {@code null}, from previous RegionObserver in the
chain
+   * @param readPoint the readpoint to create scanner
+   * @return the scanner to use during the flush.  {@code null} if the default implementation
+   * is to be used.
+   * @throws IOException if an error occurred on the coprocessor
+   */
+  InternalScanner preFlushScannerOpen(final ObserverContext<RegionCoprocessorEnvironment>
c,
+      final Store store, final KeyValueScanner memstoreScanner, final InternalScanner s,
+      final long readPoint) throws IOException;
+
+  /**
    * Called before the memstore is flushed to disk.
    * @param c the environment provided by the region server
    * @throws IOException if an error occurred on the coprocessor
@@ -283,7 +306,10 @@ public interface RegionObserver extends Coprocessor {
    * @return the scanner to use during compaction. {@code null} if the default implementation
is to
    *         be used.
    * @throws IOException if an error occurred on the coprocessor
+   * @deprecated Use {@link #preCompactScannerOpen(ObserverContext, Store, List, ScanType,
long,
+   *             InternalScanner, CompactionRequest, long)} instead.
    */
+  @Deprecated
   InternalScanner preCompactScannerOpen(final ObserverContext<RegionCoprocessorEnvironment>
c,
       final Store store, List<? extends KeyValueScanner> scanners, final ScanType scanType,
       final long earliestPutTs, final InternalScanner s, CompactionRequest request)
@@ -304,12 +330,38 @@ public interface RegionObserver extends Coprocessor {
    * @param earliestPutTs timestamp of the earliest put that was found in any of the involved
store
    *          files
    * @param s the base scanner, if not {@code null}, from previous RegionObserver in the
chain
+   * @param request compaction request
+   * @param readPoint the readpoint to create scanner
+   * @return the scanner to use during compaction. {@code null} if the default implementation
is to
+   *          be used.
+   * @throws IOException if an error occurred on the coprocessor
+   */
+  InternalScanner preCompactScannerOpen(final ObserverContext<RegionCoprocessorEnvironment>
c,
+      final Store store, List<? extends KeyValueScanner> scanners, final ScanType scanType,
+      final long earliestPutTs, final InternalScanner s, final CompactionRequest request,
+      final long readPoint) throws IOException;
+
+  /**
+   * Called prior to writing the {@link StoreFile}s selected for compaction into a new
+   * {@code StoreFile} and prior to creating the scanner used to read the input files. To
override
+   * or modify the compaction process, implementing classes can return a new scanner to provide
the
+   * KeyValues to be stored into the new {@code StoreFile} or null to perform the default
+   * processing. Calling {@link org.apache.hadoop.hbase.coprocessor.ObserverContext#bypass()}
has no
+   * effect in this hook.
+   * @param c the environment provided by the region server
+   * @param store the store being compacted
+   * @param scanners the list {@link org.apache.hadoop.hbase.regionserver.StoreFileScanner}s
+   *  to be read from
+   * @param scanType the {@link ScanType} indicating whether this is a major or minor compaction
+   * @param earliestPutTs timestamp of the earliest put that was found in any of the involved
store
+   *          files
+   * @param s the base scanner, if not {@code null}, from previous RegionObserver in the
chain
    * @return the scanner to use during compaction. {@code null} if the default implementation
is to
    *         be used.
    * @throws IOException if an error occurred on the coprocessor
    * @deprecated Use
    *             {@link #preCompactScannerOpen(ObserverContext, Store, List, ScanType, long,
-   *             InternalScanner, CompactionRequest)} instead.
+   *             InternalScanner, CompactionRequest, long)} instead.
    */
   @Deprecated
   InternalScanner preCompactScannerOpen(final ObserverContext<RegionCoprocessorEnvironment>
c,
@@ -983,9 +1035,9 @@ public interface RegionObserver extends Coprocessor {
    * Called before a store opens a new scanner.
    * This hook is called when a "user" scanner is opened.
    * <p>
-   * See {@link #preFlushScannerOpen(ObserverContext, Store, KeyValueScanner, InternalScanner)}
-   * and {@link #preCompactScannerOpen(ObserverContext,
-   *  Store, List, ScanType, long, InternalScanner)}
+   * See {@link #preFlushScannerOpen(ObserverContext, Store, KeyValueScanner, InternalScanner,
+   * long)} and {@link #preCompactScannerOpen(ObserverContext,
+   *  Store, List, ScanType, long, InternalScanner, CompactionRequest, long)}
    * to override scanners created for flushes or compactions, resp.
    * <p>
    * Call CoprocessorEnvironment#complete to skip any subsequent chained

http://git-wip-us.apache.org/repos/asf/hbase/blob/44ab659b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
index 28d2129..c564091 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
@@ -520,18 +520,19 @@ public class RegionCoprocessorHost
 
   /**
    * See
-   * {@link RegionObserver#preCompactScannerOpen(ObserverContext, Store, List, ScanType,
long, InternalScanner, CompactionRequest)}
+   * {@link RegionObserver#preCompactScannerOpen(ObserverContext, Store, List, ScanType,
long,
+   *   InternalScanner, CompactionRequest, long)}
    */
   public InternalScanner preCompactScannerOpen(final Store store,
       final List<StoreFileScanner> scanners, final ScanType scanType, final long earliestPutTs,
-      final CompactionRequest request) throws IOException {
+      final CompactionRequest request, final long readPoint) throws IOException {
     return execOperationWithResult(null,
         coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>()
{
       @Override
       public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment>
ctx)
           throws IOException {
         setResult(oserver.preCompactScannerOpen(ctx, store, scanners, scanType,
-          earliestPutTs, getResult(), request));
+          earliestPutTs, getResult(), request, readPoint));
       }
     });
   }
@@ -649,16 +650,16 @@ public class RegionCoprocessorHost
   /**
    * See
    * {@link RegionObserver#preFlushScannerOpen(ObserverContext,
-   *    Store, KeyValueScanner, InternalScanner)}
+   *    Store, KeyValueScanner, InternalScanner, long)}
    */
   public InternalScanner preFlushScannerOpen(final Store store,
-      final KeyValueScanner memstoreScanner) throws IOException {
+      final KeyValueScanner memstoreScanner, final long readPoint) throws IOException {
     return execOperationWithResult(null,
         coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>()
{
       @Override
       public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment>
ctx)
           throws IOException {
-        setResult(oserver.preFlushScannerOpen(ctx, store, memstoreScanner, getResult()));
+        setResult(oserver.preFlushScannerOpen(ctx, store, memstoreScanner, getResult(), readPoint));
       }
     });
   }

http://git-wip-us.apache.org/repos/asf/hbase/blob/44ab659b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
index 9b182a2..019e1cc 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
@@ -83,7 +83,8 @@ abstract class StoreFlusher {
       long smallestReadPoint) throws IOException {
     InternalScanner scanner = null;
     if (store.getCoprocessorHost() != null) {
-      scanner = store.getCoprocessorHost().preFlushScannerOpen(store, snapshotScanner);
+      scanner = store.getCoprocessorHost().preFlushScannerOpen(store, snapshotScanner,
+          smallestReadPoint);
     }
     if (scanner == null) {
       Scan scan = new Scan();

http://git-wip-us.apache.org/repos/asf/hbase/blob/44ab659b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
index efb5fb2..d1ab800 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
@@ -286,7 +286,8 @@ public abstract class Compactor<T extends CellSink> {
     try {
       /* Include deletes, unless we are doing a major compaction */
       ScanType scanType = scannerFactory.getScanType(request);
-      scanner = preCreateCoprocScanner(request, scanType, fd.earliestPutTs, scanners, user);
+      scanner = preCreateCoprocScanner(request, scanType, fd.earliestPutTs, scanners, user,
+        smallestReadPoint);
       if (scanner == null) {
         scanner = scannerFactory.createScanner(scanners, scanType, fd, smallestReadPoint);
       }
@@ -337,24 +338,25 @@ public abstract class Compactor<T extends CellSink> {
    * @param earliestPutTs Earliest put ts.
    * @param scanners File scanners for compaction files.
    * @param user the User
+   * @param readPoint the read point to help create scanner by Coprocessor if required.
    * @return Scanner override by coprocessor; null if not overriding.
    */
   protected InternalScanner preCreateCoprocScanner(final CompactionRequest request,
       final ScanType scanType, final long earliestPutTs, final List<StoreFileScanner>
scanners,
-      User user) throws IOException {
+      User user, final long readPoint) throws IOException {
     if (store.getCoprocessorHost() == null) {
       return null;
     }
     if (user == null) {
       return store.getCoprocessorHost().preCompactScannerOpen(store, scanners, scanType,
-        earliestPutTs, request);
+        earliestPutTs, request, readPoint);
     } else {
       try {
         return user.getUGI().doAs(new PrivilegedExceptionAction<InternalScanner>()
{
           @Override
           public InternalScanner run() throws Exception {
             return store.getCoprocessorHost().preCompactScannerOpen(store, scanners,
-              scanType, earliestPutTs, request);
+              scanType, earliestPutTs, request, readPoint);
           }
         });
       } catch (InterruptedException ie) {


Mime
View raw message