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 java.util.List;
021
022import org.apache.hadoop.hbase.Cell;
023import org.apache.hadoop.hbase.HBaseInterfaceAudience;
024import org.apache.yetus.audience.InterfaceAudience;
025import org.apache.yetus.audience.InterfaceStability;
026import org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics;
027
028/**
029 * ScannerContext instances encapsulate limit tracking AND progress towards those limits during
030 * invocations of {@link InternalScanner#next(java.util.List)} and
031 * {@link RegionScanner#next(java.util.List)}.
032 * <p>
033 * A ScannerContext instance should be updated periodically throughout execution whenever progress
034 * towards a limit has been made. Each limit can be checked via the appropriate checkLimit method.
035 * <p>
036 * Once a limit has been reached, the scan will stop. The invoker of
037 * {@link InternalScanner#next(java.util.List)} or {@link RegionScanner#next(java.util.List)} can
038 * use the appropriate check*Limit methods to see exactly which limits have been reached.
039 * Alternatively, {@link #checkAnyLimitReached(LimitScope)} is provided to see if ANY limit was
040 * reached
041 * <p>
042 * {@link NoLimitScannerContext#NO_LIMIT} is an immutable static definition that can be used
043 * whenever a {@link ScannerContext} is needed but limits do not need to be enforced.
044 * <p>
045 * NOTE: It is important that this class only ever expose setter methods that can be safely skipped
046 * when limits should be NOT enforced. This is because of the necessary immutability of the class
047 * {@link NoLimitScannerContext}. If a setter cannot be safely skipped, the immutable nature of
048 * {@link NoLimitScannerContext} will lead to incorrect behavior.
049 */
050@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
051@InterfaceStability.Evolving
052public class ScannerContext {
053
054  LimitFields limits;
055  /**
056   * A different set of progress fields. Only include batch, dataSize and heapSize. Compare to
057   * LimitFields, ProgressFields doesn't contain time field. As we save a deadline in LimitFields,
058   * so use {@link System#currentTimeMillis()} directly when check time limit.
059   */
060  ProgressFields progress;
061
062  /**
063   * The state of the scanner after the invocation of {@link InternalScanner#next(java.util.List)}
064   * or {@link RegionScanner#next(java.util.List)}.
065   */
066  NextState scannerState;
067  private static final NextState DEFAULT_STATE = NextState.MORE_VALUES;
068
069  /**
070   * Used as an indication to invocations of {@link InternalScanner#next(java.util.List)} and
071   * {@link RegionScanner#next(java.util.List)} that, if true, the progress tracked within this
072   * {@link ScannerContext} instance should be considered while evaluating the limits. Useful for
073   * enforcing a set of limits across multiple calls (i.e. the limit may not be reached in a single
074   * invocation, but any progress made should be considered in future invocations)
075   * <p>
076   * Defaulting this value to false means that, by default, any tracked progress will be wiped clean
077   * on invocations to {@link InternalScanner#next(java.util.List)} and
078   * {@link RegionScanner#next(java.util.List)} and the call will be treated as though no progress
079   * has been made towards the limits so far.
080   * <p>
081   * This is an important mechanism. Users of Internal/Region scanners expect that they can define
082   * some limits and then repeatedly invoke {@link InternalScanner#next(List)} or
083   * {@link RegionScanner#next(List)} where each invocation respects these limits separately.
084   * <p>
085   * For example: <pre> {@code
086   * ScannerContext context = new ScannerContext.newBuilder().setBatchLimit(5).build();
087   * RegionScanner scanner = ...
088   * List<Cell> results = new ArrayList<Cell>();
089   * while(scanner.next(results, context)) {
090   *   // Do something with a batch of 5 cells
091   * }
092   * }</pre> However, in the case of RPCs, the server wants to be able to define a set of
093   * limits for a particular RPC request and have those limits respected across multiple
094   * invocations. This means that the progress made towards the limits in earlier calls will be
095   * saved and considered in future invocations
096   */
097  boolean keepProgress;
098  private static boolean DEFAULT_KEEP_PROGRESS = false;
099
100  private Cell lastPeekedCell = null;
101
102  // Set this to true will have the same behavior with reaching the time limit.
103  // This is used when you want to make the current RSRpcService.scan returns immediately. For
104  // example, when we want to switch from pread to stream, we can only do it after the rpc call is
105  // returned.
106  private boolean returnImmediately;
107
108  /**
109   * Tracks the relevant server side metrics during scans. null when metrics should not be tracked
110   */
111  final ServerSideScanMetrics metrics;
112
113  ScannerContext(boolean keepProgress, LimitFields limitsToCopy, boolean trackMetrics) {
114    this.limits = new LimitFields();
115    if (limitsToCopy != null) {
116      this.limits.copy(limitsToCopy);
117    }
118
119    // Progress fields are initialized to 0
120    progress = new ProgressFields(0, 0, 0);
121
122    this.keepProgress = keepProgress;
123    this.scannerState = DEFAULT_STATE;
124    this.metrics = trackMetrics ? new ServerSideScanMetrics() : null;
125  }
126
127  public boolean isTrackingMetrics() {
128    return this.metrics != null;
129  }
130
131  /**
132   * Get the metrics instance. Should only be called after a call to {@link #isTrackingMetrics()}
133   * has been made to confirm that metrics are indeed being tracked.
134   * @return {@link ServerSideScanMetrics} instance that is tracking metrics for this scan
135   */
136  public ServerSideScanMetrics getMetrics() {
137    assert isTrackingMetrics();
138    return this.metrics;
139  }
140
141  /**
142   * @return true if the progress tracked so far in this instance will be considered during an
143   *         invocation of {@link InternalScanner#next(java.util.List)} or
144   *         {@link RegionScanner#next(java.util.List)}. false when the progress tracked so far
145   *         should not be considered and should instead be wiped away via {@link #clearProgress()}
146   */
147  boolean getKeepProgress() {
148    return keepProgress;
149  }
150
151  void setKeepProgress(boolean keepProgress) {
152    this.keepProgress = keepProgress;
153  }
154
155  /**
156   * Progress towards the batch limit has been made. Increment internal tracking of batch progress
157   */
158  void incrementBatchProgress(int batch) {
159    int currentBatch = progress.getBatch();
160    progress.setBatch(currentBatch + batch);
161  }
162
163  /**
164   * Progress towards the size limit has been made. Increment internal tracking of size progress
165   */
166  void incrementSizeProgress(long dataSize, long heapSize) {
167    long curDataSize = progress.getDataSize();
168    progress.setDataSize(curDataSize + dataSize);
169    long curHeapSize = progress.getHeapSize();
170    progress.setHeapSize(curHeapSize + heapSize);
171  }
172
173  int getBatchProgress() {
174    return progress.getBatch();
175  }
176
177  long getDataSizeProgress() {
178    return progress.getDataSize();
179  }
180
181  long getHeapSizeProgress() {
182    return progress.getHeapSize();
183  }
184
185  void setProgress(int batchProgress, long sizeProgress, long heapSizeProgress) {
186    setBatchProgress(batchProgress);
187    setSizeProgress(sizeProgress, heapSizeProgress);
188  }
189
190  void setSizeProgress(long dataSizeProgress, long heapSizeProgress) {
191    progress.setDataSize(dataSizeProgress);
192    progress.setHeapSize(heapSizeProgress);
193  }
194
195  void setBatchProgress(int batchProgress) {
196    progress.setBatch(batchProgress);
197  }
198
199  /**
200   * Clear away any progress that has been made so far. All progress fields are reset to initial
201   * values
202   */
203  void clearProgress() {
204    progress.setFields(0, 0, 0);
205  }
206
207  /**
208   * Note that this is not a typical setter. This setter returns the {@link NextState} that was
209   * passed in so that methods can be invoked against the new state. Furthermore, this pattern
210   * allows the {@link NoLimitScannerContext} to cleanly override this setter and simply return the
211   * new state, thus preserving the immutability of {@link NoLimitScannerContext}
212   * @param state
213   * @return The state that was passed in.
214   */
215  NextState setScannerState(NextState state) {
216    if (!NextState.isValidState(state)) {
217      throw new IllegalArgumentException("Cannot set to invalid state: " + state);
218    }
219
220    this.scannerState = state;
221    return state;
222  }
223
224  /**
225   * @return true when we have more cells for the current row. This usually because we have reached
226   *         a limit in the middle of a row
227   */
228  boolean mayHaveMoreCellsInRow() {
229    return scannerState == NextState.SIZE_LIMIT_REACHED_MID_ROW ||
230        scannerState == NextState.TIME_LIMIT_REACHED_MID_ROW ||
231        scannerState == NextState.BATCH_LIMIT_REACHED;
232  }
233
234  /**
235   * @param checkerScope
236   * @return true if the batch limit can be enforced in the checker's scope
237   */
238  boolean hasBatchLimit(LimitScope checkerScope) {
239    return limits.canEnforceBatchLimitFromScope(checkerScope) && limits.getBatch() > 0;
240  }
241
242  /**
243   * @param checkerScope
244   * @return true if the size limit can be enforced in the checker's scope
245   */
246  boolean hasSizeLimit(LimitScope checkerScope) {
247    return limits.canEnforceSizeLimitFromScope(checkerScope)
248        && (limits.getDataSize() > 0 || limits.getHeapSize() > 0);
249  }
250
251  /**
252   * @param checkerScope
253   * @return true if the time limit can be enforced in the checker's scope
254   */
255  boolean hasTimeLimit(LimitScope checkerScope) {
256    return limits.canEnforceTimeLimitFromScope(checkerScope) &&
257      (limits.getTime() > 0 || returnImmediately);
258  }
259
260  /**
261   * @param checkerScope
262   * @return true if any limit can be enforced within the checker's scope
263   */
264  boolean hasAnyLimit(LimitScope checkerScope) {
265    return hasBatchLimit(checkerScope) || hasSizeLimit(checkerScope) || hasTimeLimit(checkerScope);
266  }
267
268  /**
269   * @param scope The scope in which the size limit will be enforced
270   */
271  void setSizeLimitScope(LimitScope scope) {
272    limits.setSizeScope(scope);
273  }
274
275  /**
276   * @param scope The scope in which the time limit will be enforced
277   */
278  void setTimeLimitScope(LimitScope scope) {
279    limits.setTimeScope(scope);
280  }
281
282  int getBatchLimit() {
283    return limits.getBatch();
284  }
285
286  long getDataSizeLimit() {
287    return limits.getDataSize();
288  }
289
290  long getTimeLimit() {
291    return limits.getTime();
292  }
293
294  /**
295   * @param checkerScope The scope that the limit is being checked from
296   * @return true when the limit is enforceable from the checker's scope and it has been reached
297   */
298  boolean checkBatchLimit(LimitScope checkerScope) {
299    return hasBatchLimit(checkerScope) && progress.getBatch() >= limits.getBatch();
300  }
301
302  /**
303   * @param checkerScope The scope that the limit is being checked from
304   * @return true when the limit is enforceable from the checker's scope and it has been reached
305   */
306  boolean checkSizeLimit(LimitScope checkerScope) {
307    return hasSizeLimit(checkerScope) && (progress.getDataSize() >= limits.getDataSize()
308        || progress.getHeapSize() >= limits.getHeapSize());
309  }
310
311  /**
312   * @param checkerScope The scope that the limit is being checked from. The time limit is always
313   *          checked against {@link System#currentTimeMillis()}
314   * @return true when the limit is enforceable from the checker's scope and it has been reached
315   */
316  boolean checkTimeLimit(LimitScope checkerScope) {
317    return hasTimeLimit(checkerScope) &&
318      (returnImmediately || System.currentTimeMillis() >= limits.getTime());
319  }
320
321  /**
322   * @param checkerScope The scope that the limits are being checked from
323   * @return true when some limit is enforceable from the checker's scope and it has been reached
324   */
325  boolean checkAnyLimitReached(LimitScope checkerScope) {
326    return checkSizeLimit(checkerScope) || checkBatchLimit(checkerScope)
327        || checkTimeLimit(checkerScope);
328  }
329
330  Cell getLastPeekedCell() {
331    return lastPeekedCell;
332  }
333
334  void setLastPeekedCell(Cell lastPeekedCell) {
335    this.lastPeekedCell = lastPeekedCell;
336  }
337
338  void returnImmediately() {
339    this.returnImmediately = true;
340  }
341
342  @Override
343  public String toString() {
344    StringBuilder sb = new StringBuilder();
345    sb.append("{");
346
347    sb.append("limits:");
348    sb.append(limits);
349
350    sb.append(", progress:");
351    sb.append(progress);
352
353    sb.append(", keepProgress:");
354    sb.append(keepProgress);
355
356    sb.append(", state:");
357    sb.append(scannerState);
358
359    sb.append("}");
360    return sb.toString();
361  }
362
363  public static Builder newBuilder() {
364    return new Builder();
365  }
366
367  public static Builder newBuilder(boolean keepProgress) {
368    return new Builder(keepProgress);
369  }
370
371  public static final class Builder {
372    boolean keepProgress = DEFAULT_KEEP_PROGRESS;
373    boolean trackMetrics = false;
374    LimitFields limits = new LimitFields();
375
376    private Builder() {
377    }
378
379    private Builder(boolean keepProgress) {
380      this.keepProgress = keepProgress;
381    }
382
383    public Builder setKeepProgress(boolean keepProgress) {
384      this.keepProgress = keepProgress;
385      return this;
386    }
387
388    public Builder setTrackMetrics(boolean trackMetrics) {
389      this.trackMetrics = trackMetrics;
390      return this;
391    }
392
393    public Builder setSizeLimit(LimitScope sizeScope, long dataSizeLimit, long heapSizeLimit) {
394      limits.setDataSize(dataSizeLimit);
395      limits.setHeapSize(heapSizeLimit);
396      limits.setSizeScope(sizeScope);
397      return this;
398    }
399
400    public Builder setTimeLimit(LimitScope timeScope, long timeLimit) {
401      limits.setTime(timeLimit);
402      limits.setTimeScope(timeScope);
403      return this;
404    }
405
406    public Builder setBatchLimit(int batchLimit) {
407      limits.setBatch(batchLimit);
408      return this;
409    }
410
411    public ScannerContext build() {
412      return new ScannerContext(keepProgress, limits, trackMetrics);
413    }
414  }
415
416  /**
417   * The possible states a scanner may be in following a call to {@link InternalScanner#next(List)}
418   */
419  public enum NextState {
420    MORE_VALUES(true, false),
421    NO_MORE_VALUES(false, false),
422    SIZE_LIMIT_REACHED(true, true),
423
424    /**
425     * Special case of size limit reached to indicate that the size limit was reached in the middle
426     * of a row and thus a partial results was formed
427     */
428    SIZE_LIMIT_REACHED_MID_ROW(true, true),
429    TIME_LIMIT_REACHED(true, true),
430
431    /**
432     * Special case of time limit reached to indicate that the time limit was reached in the middle
433     * of a row and thus a partial results was formed
434     */
435    TIME_LIMIT_REACHED_MID_ROW(true, true),
436    BATCH_LIMIT_REACHED(true, true);
437
438    private final boolean moreValues;
439    private final boolean limitReached;
440
441    private NextState(boolean moreValues, boolean limitReached) {
442      this.moreValues = moreValues;
443      this.limitReached = limitReached;
444    }
445
446    /**
447     * @return true when the state indicates that more values may follow those that have been
448     *         returned
449     */
450    public boolean hasMoreValues() {
451      return this.moreValues;
452    }
453
454    /**
455     * @return true when the state indicates that a limit has been reached and scan should stop
456     */
457    public boolean limitReached() {
458      return this.limitReached;
459    }
460
461    public static boolean isValidState(NextState state) {
462      return state != null;
463    }
464
465    public static boolean hasMoreValues(NextState state) {
466      return isValidState(state) && state.hasMoreValues();
467    }
468  }
469
470  /**
471   * The various scopes where a limit can be enforced. Used to differentiate when a limit should be
472   * enforced or not.
473   */
474  public enum LimitScope {
475    /**
476     * Enforcing a limit between rows means that the limit will not be considered until all the
477     * cells for a particular row have been retrieved
478     */
479    BETWEEN_ROWS(0),
480
481    /**
482     * Enforcing a limit between cells means that the limit will be considered after each full cell
483     * has been retrieved
484     */
485    BETWEEN_CELLS(1);
486
487    /**
488     * When enforcing a limit, we must check that the scope is appropriate for enforcement.
489     * <p>
490     * To communicate this concept, each scope has a depth. A limit will be enforced if the depth of
491     * the checker's scope is less than or equal to the limit's scope. This means that when checking
492     * limits, the checker must know their own scope (i.e. are they checking the limits between
493     * rows, between cells, etc...)
494     */
495    final int depth;
496
497    LimitScope(int depth) {
498      this.depth = depth;
499    }
500
501    final int depth() {
502      return depth;
503    }
504
505    /**
506     * @param checkerScope The scope in which the limit is being checked
507     * @return true when the checker is in a scope that indicates the limit can be enforced. Limits
508     *         can be enforced from "higher or equal" scopes (i.e. the checker's scope is at a
509     *         lesser depth than the limit)
510     */
511    boolean canEnforceLimitFromScope(LimitScope checkerScope) {
512      return checkerScope != null && checkerScope.depth() <= depth;
513    }
514  }
515
516  /**
517   * The different fields that can be used as limits in calls to
518   * {@link InternalScanner#next(java.util.List)} and {@link RegionScanner#next(java.util.List)}
519   */
520  private static class LimitFields {
521    /**
522     * Default values of the limit fields. Defined such that if a field does NOT change from its
523     * default, it will not be enforced
524     */
525    private static int DEFAULT_BATCH = -1;
526    private static long DEFAULT_SIZE = -1L;
527    private static long DEFAULT_TIME = -1L;
528
529    /**
530     * Default scope that is assigned to a limit if a scope is not specified.
531     */
532    private static final LimitScope DEFAULT_SCOPE = LimitScope.BETWEEN_ROWS;
533
534    // The batch limit will always be enforced between cells, thus, there isn't a field to hold the
535    // batch scope
536    int batch = DEFAULT_BATCH;
537
538    LimitScope sizeScope = DEFAULT_SCOPE;
539    // The sum of cell data sizes(key + value). The Cell data might be in on heap or off heap area.
540    long dataSize = DEFAULT_SIZE;
541    // The sum of heap space occupied by all tracked cells. This includes Cell POJO's overhead as
542    // such AND data cells of Cells which are in on heap area.
543    long heapSize = DEFAULT_SIZE;
544
545    LimitScope timeScope = DEFAULT_SCOPE;
546    long time = DEFAULT_TIME;
547
548    /**
549     * Fields keep their default values.
550     */
551    LimitFields() {
552    }
553
554    void copy(LimitFields limitsToCopy) {
555      if (limitsToCopy != null) {
556        setFields(limitsToCopy.getBatch(), limitsToCopy.getSizeScope(), limitsToCopy.getDataSize(),
557            limitsToCopy.getHeapSize(), limitsToCopy.getTimeScope(), limitsToCopy.getTime());
558      }
559    }
560
561    /**
562     * Set all fields together.
563     */
564    void setFields(int batch, LimitScope sizeScope, long dataSize, long heapSize,
565        LimitScope timeScope, long time) {
566      setBatch(batch);
567      setSizeScope(sizeScope);
568      setDataSize(dataSize);
569      setHeapSize(heapSize);
570      setTimeScope(timeScope);
571      setTime(time);
572    }
573
574    int getBatch() {
575      return this.batch;
576    }
577
578    void setBatch(int batch) {
579      this.batch = batch;
580    }
581
582    /**
583     * @param checkerScope
584     * @return true when the limit can be enforced from the scope of the checker
585     */
586    boolean canEnforceBatchLimitFromScope(LimitScope checkerScope) {
587      return LimitScope.BETWEEN_CELLS.canEnforceLimitFromScope(checkerScope);
588    }
589
590    long getDataSize() {
591      return this.dataSize;
592    }
593
594    long getHeapSize() {
595      return this.heapSize;
596    }
597
598    void setDataSize(long dataSize) {
599      this.dataSize = dataSize;
600    }
601
602    void setHeapSize(long heapSize) {
603      this.heapSize = heapSize;
604    }
605
606    /**
607     * @return {@link LimitScope} indicating scope in which the size limit is enforced
608     */
609    LimitScope getSizeScope() {
610      return this.sizeScope;
611    }
612
613    /**
614     * Change the scope in which the size limit is enforced
615     */
616    void setSizeScope(LimitScope scope) {
617      this.sizeScope = scope;
618    }
619
620    /**
621     * @param checkerScope
622     * @return true when the limit can be enforced from the scope of the checker
623     */
624    boolean canEnforceSizeLimitFromScope(LimitScope checkerScope) {
625      return this.sizeScope.canEnforceLimitFromScope(checkerScope);
626    }
627
628    long getTime() {
629      return this.time;
630    }
631
632    void setTime(long time) {
633      this.time = time;
634    }
635
636    /**
637     * @return {@link LimitScope} indicating scope in which the time limit is enforced
638     */
639    LimitScope getTimeScope() {
640      return this.timeScope;
641    }
642
643    /**
644     * Change the scope in which the time limit is enforced
645     */
646    void setTimeScope(LimitScope scope) {
647      this.timeScope = scope;
648    }
649
650    /**
651     * @param checkerScope
652     * @return true when the limit can be enforced from the scope of the checker
653     */
654    boolean canEnforceTimeLimitFromScope(LimitScope checkerScope) {
655      return this.timeScope.canEnforceLimitFromScope(checkerScope);
656    }
657
658    @Override
659    public String toString() {
660      StringBuilder sb = new StringBuilder();
661      sb.append("{");
662
663      sb.append("batch:");
664      sb.append(batch);
665
666      sb.append(", dataSize:");
667      sb.append(dataSize);
668
669      sb.append(", heapSize:");
670      sb.append(heapSize);
671
672      sb.append(", sizeScope:");
673      sb.append(sizeScope);
674
675      sb.append(", time:");
676      sb.append(time);
677
678      sb.append(", timeScope:");
679      sb.append(timeScope);
680
681      sb.append("}");
682      return sb.toString();
683    }
684  }
685
686  private static class ProgressFields {
687
688    private static int DEFAULT_BATCH = -1;
689    private static long DEFAULT_SIZE = -1L;
690
691    // The batch limit will always be enforced between cells, thus, there isn't a field to hold the
692    // batch scope
693    int batch = DEFAULT_BATCH;
694
695    // The sum of cell data sizes(key + value). The Cell data might be in on heap or off heap area.
696    long dataSize = DEFAULT_SIZE;
697    // The sum of heap space occupied by all tracked cells. This includes Cell POJO's overhead as
698    // such AND data cells of Cells which are in on heap area.
699    long heapSize = DEFAULT_SIZE;
700
701    ProgressFields(int batch, long size, long heapSize) {
702      setFields(batch, size, heapSize);
703    }
704
705    /**
706     * Set all fields together.
707     */
708    void setFields(int batch, long dataSize, long heapSize) {
709      setBatch(batch);
710      setDataSize(dataSize);
711      setHeapSize(heapSize);
712    }
713
714    int getBatch() {
715      return this.batch;
716    }
717
718    void setBatch(int batch) {
719      this.batch = batch;
720    }
721
722    long getDataSize() {
723      return this.dataSize;
724    }
725
726    long getHeapSize() {
727      return this.heapSize;
728    }
729
730    void setDataSize(long dataSize) {
731      this.dataSize = dataSize;
732    }
733
734    void setHeapSize(long heapSize) {
735      this.heapSize = heapSize;
736    }
737
738    @Override
739    public String toString() {
740      StringBuilder sb = new StringBuilder();
741      sb.append("{");
742
743      sb.append("batch:");
744      sb.append(batch);
745
746      sb.append(", dataSize:");
747      sb.append(dataSize);
748
749      sb.append(", heapSize:");
750      sb.append(heapSize);
751
752      sb.append("}");
753      return sb.toString();
754    }
755  }
756}