hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yangzhe1...@apache.org
Subject [3/5] hbase git commit: HBASE-15576 Scanning cursor to prevent blocking long time on ResultScanner.next()
Date Wed, 07 Jun 2017 03:38:38 GMT
HBASE-15576 Scanning cursor to prevent blocking long time on ResultScanner.next()


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

Branch: refs/heads/master
Commit: 2f1923a8233b0c999494cd4b33f85b70dc5d7b12
Parents: 80e15aa
Author: Phil Yang <yangzhe1991@apache.org>
Authored: Thu May 25 15:18:58 2017 +0800
Committer: Phil Yang <yangzhe1991@apache.org>
Committed: Wed Jun 7 11:32:04 2017 +0800

----------------------------------------------------------------------
 .../hadoop/hbase/client/ClientScanner.java      |   15 +
 .../apache/hadoop/hbase/client/ClientUtil.java  |    4 +
 .../org/apache/hadoop/hbase/client/Cursor.java  |   41 +
 .../org/apache/hadoop/hbase/client/Result.java  |   45 +
 .../org/apache/hadoop/hbase/client/Scan.java    |   43 +
 .../hadoop/hbase/client/ScannerCallable.java    |   14 +-
 .../client/ScannerCallableWithReplicas.java     |    4 +
 .../hbase/shaded/protobuf/ProtobufUtil.java     |   24 +
 .../shaded/protobuf/generated/ClientProtos.java | 1172 +++++++++++++++---
 .../src/main/protobuf/Client.proto              |   14 +-
 .../hbase/protobuf/generated/ClientProtos.java  | 1128 ++++++++++++++---
 hbase-protocol/src/main/protobuf/Client.proto   |   14 +-
 .../hbase/regionserver/RSRpcServices.java       |   18 +-
 .../hbase/regionserver/ScannerContext.java      |   13 +
 .../hadoop/hbase/regionserver/StoreScanner.java |    1 +
 .../hbase/regionserver/TestScannerCursor.java   |  191 +++
 16 files changed, 2422 insertions(+), 319 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
index fa5f868..59cf005 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
@@ -499,6 +499,21 @@ public abstract class ClientScanner extends AbstractClientScanner {
           break;
         }
       }
+      if (cache.isEmpty() && !closed && scan.isNeedCursorResult()) {
+        if (callable.isHeartbeatMessage() && callable.getCursor() != null) {
+          // Use cursor row key from server
+          cache.add(Result.createCursorResult(callable.getCursor()));
+          break;
+        }
+        if (values.length > 0) {
+          // It is size limit exceed and we need return the last Result's row.
+          // When user setBatch and the scanner is reopened, the server may return Results
that
+          // user has seen and the last Result can not be seen because the number is not
enough.
+          // So the row keys of results may not be same, we must use the last one.
+          cache.add(Result.createCursorResult(new Cursor(values[values.length - 1].getRow())));
+          break;
+        }
+      }
       if (countdown <= 0) {
         // we have enough result.
         closeScannerIfExhausted(regionExhausted);

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientUtil.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientUtil.java
index e4a84d5..a839080 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientUtil.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientUtil.java
@@ -27,4 +27,8 @@ public class ClientUtil {
   public static boolean areScanStartRowAndStopRowEqual(byte[] startRow, byte[] stopRow) {
     return startRow != null && startRow.length > 0 && Bytes.equals(startRow,
stopRow);
   }
+
+  public static Cursor createCursor(byte[] row) {
+    return new Cursor(row);
+  }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Cursor.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Cursor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Cursor.java
new file mode 100644
index 0000000..1d4b4b5
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Cursor.java
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.client;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+/**
+ * Scan cursor to tell client where server is scanning
+ * {@link Scan#setNeedCursorResult(boolean)}
+ * {@link Result#isCursor()}
+ * {@link Result#getCursor()}
+ */
+@InterfaceAudience.Public
+public class Cursor {
+
+  private final byte[] row;
+
+  Cursor(byte[] row) {
+    this.row = row;
+  }
+
+  public byte[] getRow() {
+    return row;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
index 63aab80..94e1b90 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
@@ -108,6 +108,8 @@ public class Result implements CellScannable, CellScanner {
 
   private final boolean readonly;
 
+  private Cursor cursor = null;
+
   /**
    * Creates an empty Result w/ no KeyValue payload; returns null if you call {@link #rawCells()}.
    * Use this to represent no results if {@code null} won't do or in old 'mapred' as opposed
@@ -173,6 +175,15 @@ public class Result implements CellScannable, CellScanner {
     return new Result(cells, null, stale, mayHaveMoreCellsInRow);
   }
 
+  public static Result createCursorResult(Cursor cursor) {
+    return new Result(cursor);
+  }
+
+  private Result(Cursor cursor) {
+    this.cursor = cursor;
+    this.readonly = false;
+  }
+
   /** Private ctor. Use {@link #create(Cell[])}. */
   private Result(Cell[] cells, Boolean exists, boolean stale, boolean mayHaveMoreCellsInRow)
{
     this.cells = cells;
@@ -948,4 +959,38 @@ public class Result implements CellScannable, CellScanner {
       throw new UnsupportedOperationException("Attempting to modify readonly EMPTY_RESULT!");
     }
   }
+
+  /**
+   * Return true if this Result is a cursor to tell users where the server has scanned.
+   * In this Result the only meaningful method is {@link #getCursor()}.
+   *
+   * {@code
+   *  while (r = scanner.next() && r != null) {
+   *    if(r.isCursor()){
+   *    // scanning is not end, it is a cursor, save its row key and close scanner if you
want, or
+   *    // just continue the loop to call next().
+   *    } else {
+   *    // just like before
+   *    }
+   *  }
+   *  // scanning is end
+   *
+   * }
+   * {@link Scan#setNeedCursorResult(boolean)}
+   * {@link Cursor}
+   * {@link #getCursor()}
+   */
+  public boolean isCursor() {
+    return cursor != null ;
+  }
+
+  /**
+   * Return the cursor if this Result is a cursor result.
+   * {@link Scan#setNeedCursorResult(boolean)}
+   * {@link Cursor}
+   * {@link #isCursor()}
+   */
+  public Cursor getCursor(){
+    return cursor;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
index 2746263..639f43e 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Scan.java
@@ -185,6 +185,9 @@ public class Scan extends Query {
    * Control whether to use pread at server side.
    */
   private ReadType readType = ReadType.DEFAULT;
+
+  private boolean needCursorResult = false;
+
   /**
    * Create a Scan operation across all rows.
    */
@@ -272,6 +275,7 @@ public class Scan extends Query {
     }
     this.mvccReadPoint = scan.getMvccReadPoint();
     this.limit = scan.getLimit();
+    this.needCursorResult = scan.isNeedCursorResult();
   }
 
   /**
@@ -1170,4 +1174,43 @@ public class Scan extends Query {
   Scan resetMvccReadPoint() {
     return setMvccReadPoint(-1L);
   }
+
+  /**
+   * When the server is slow or we scan a table with many deleted data or we use a sparse
filter,
+   * the server will response heartbeat to prevent timeout. However the scanner will return
a Result
+   * only when client can do it. So if there are many heartbeats, the blocking time on
+   * ResultScanner#next() may be very long, which is not friendly to online services.
+   *
+   * Set this to true then you can get a special Result whose #isCursor() returns true and
is not
+   * contains any real data. It only tells you where the server has scanned. You can call
next
+   * to continue scanning or open a new scanner with this row key as start row whenever you
want.
+   *
+   * Users can get a cursor when and only when there is a response from the server but we
can not
+   * return a Result to users, for example, this response is a heartbeat or there are partial
cells
+   * but users do not allow partial result.
+   *
+   * Now the cursor is in row level which means the special Result will only contains a row
key.
+   * {@link Result#isCursor()}
+   * {@link Result#getCursor()}
+   * {@link Cursor}
+   */
+  public Scan setNeedCursorResult(boolean needCursorResult) {
+    this.needCursorResult = needCursorResult;
+    return this;
+  }
+
+  public boolean isNeedCursorResult() {
+    return needCursorResult;
+  }
+
+  /**
+   * Create a new Scan with a cursor. It only set the position information like start row
key.
+   * The others (like cfs, stop row, limit) should still be filled in by the user.
+   * {@link Result#isCursor()}
+   * {@link Result#getCursor()}
+   * {@link Cursor}
+   */
+  public static Scan createScanFromCursor(Cursor cursor) {
+    return new Scan().withStartRow(cursor.getRow());
+  }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
index ffac566..4227e41 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
@@ -87,6 +87,8 @@ public class ScannerCallable extends ClientServiceCallable<Result[]>
{
    */
   protected boolean heartbeatMessage = false;
 
+  protected Cursor cursor;
+
   // indicate if it is a remote server call
   protected boolean isRegionServerRemote = true;
   private long nextCallSeq = 0;
@@ -148,7 +150,7 @@ public class ScannerCallable extends ClientServiceCallable<Result[]>
{
       checkIfRegionServerIsRemote();
       instantiated = true;
     }
-
+    cursor = null;
     // check how often we retry.
     if (reload) {
       incRPCRetriesMetrics(scanMetrics, isRegionServerRemote);
@@ -242,7 +244,11 @@ public class ScannerCallable extends ClientServiceCallable<Result[]>
{
       response = next();
     }
     long timestamp = System.currentTimeMillis();
-    setHeartbeatMessage(response.hasHeartbeatMessage() && response.getHeartbeatMessage());
+    boolean isHeartBeat = response.hasHeartbeatMessage() && response.getHeartbeatMessage();
+    setHeartbeatMessage(isHeartBeat);
+    if (isHeartBeat && scan.isNeedCursorResult() && response.hasCursor())
{
+      cursor = ProtobufUtil.toCursor(response.getCursor());
+    }
     Result[] rrs = ResponseConverter.getResults(getRpcControllerCellScanner(), response);
     if (logScannerActivity) {
       long now = System.currentTimeMillis();
@@ -288,6 +294,10 @@ public class ScannerCallable extends ClientServiceCallable<Result[]>
{
     return heartbeatMessage;
   }
 
+  public Cursor getCursor() {
+    return cursor;
+  }
+
   private void setHeartbeatMessage(boolean heartbeatMessage) {
     this.heartbeatMessage = heartbeatMessage;
   }

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallableWithReplicas.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallableWithReplicas.java
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallableWithReplicas.java
index bcd5d21..0cdd4dd 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallableWithReplicas.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallableWithReplicas.java
@@ -302,6 +302,10 @@ class ScannerCallableWithReplicas implements RetryingCallable<Result[]>
{
     return currentScannerCallable != null && currentScannerCallable.isHeartbeatMessage();
   }
 
+  public Cursor getCursor() {
+    return currentScannerCallable != null ? currentScannerCallable.getCursor() : null;
+  }
+
   private void addCallsForCurrentReplica(
       ResultBoundedCompletionService<Pair<Result[], ScannerCallable>> cs, RegionLocations
rl) {
     RetryingRPC retryingOnReplica = new RetryingRPC(currentScannerCallable);

http://git-wip-us.apache.org/repos/asf/hbase/blob/2f1923a8/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
index 8a4e412..5c4dd55 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
@@ -65,6 +65,7 @@ import org.apache.hadoop.hbase.client.Append;
 import org.apache.hadoop.hbase.client.ClientUtil;
 import org.apache.hadoop.hbase.client.CompactionState;
 import org.apache.hadoop.hbase.client.Consistency;
+import org.apache.hadoop.hbase.client.Cursor;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Durability;
 import org.apache.hadoop.hbase.client.Get;
@@ -185,6 +186,7 @@ import org.apache.hadoop.hbase.util.Methods;
 import org.apache.hadoop.hbase.util.NonceKey;
 import org.apache.hadoop.hbase.util.VersionInfo;
 import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hadoop.mapreduce.tools.CLI;
 
 /**
  * Protobufs utility.
@@ -1099,6 +1101,9 @@ public final class ProtobufUtil {
     if (scan.getReadType() != Scan.ReadType.DEFAULT) {
       scanBuilder.setReadType(toReadType(scan.getReadType()));
     }
+    if (scan.isNeedCursorResult()) {
+      scanBuilder.setNeedCursorResult(true);
+    }
     return scanBuilder.build();
   }
 
@@ -1207,9 +1212,28 @@ public final class ProtobufUtil {
     } else if (proto.hasReadType()) {
       scan.setReadType(toReadType(proto.getReadType()));
     }
+    if (proto.getNeedCursorResult()) {
+      scan.setNeedCursorResult(true);
+    }
     return scan;
   }
 
+  public static ClientProtos.Cursor toCursor(Cursor cursor) {
+    ClientProtos.Cursor.Builder builder = ClientProtos.Cursor.newBuilder();
+    ClientProtos.Cursor.newBuilder().setRow(ByteString.copyFrom(cursor.getRow()));
+    return builder.build();
+  }
+
+  public static ClientProtos.Cursor toCursor(Cell cell) {
+    return ClientProtos.Cursor.newBuilder()
+        .setRow(ByteString.copyFrom(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()))
+        .build();
+  }
+
+  public static Cursor toCursor(ClientProtos.Cursor cursor) {
+    return ClientUtil.createCursor(cursor.getRow().toByteArray());
+  }
+
   /**
    * Create a protocol buffer Get based on a client Get.
    *


Mime
View raw message