HBASE-25975: Row Commit Sequencer

Use a row commit sequencer in HRegion to ensure that only the operations
that mutate disjoint sets of rows are able to commit within the same clock
tick. This maintains the invariant that more than one mutation to a given
row will never be committed in the same clock tick.

Callers will first acquire row locks for the row(s) the pending mutation
will mutate. Then they will use RowCommitSequencer.getRowSequence to
ensure that the set of rows about to be mutated do not overlap with those
for any other pending mutations in the current clock tick. If an overlap
is identified, getRowSequence will yield and loop until there is no
longer an overlap and the caller's pending mutation can succeed.

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionSource.java

@@ -38,6 +38,7 @@ public interface MetricsRegionSource extends Comparable<MetricsRegionSource> {
   String NUM_FILES_COMPACTED_COUNT = "numFilesCompactedCount";
   String FLUSHES_QUEUED_COUNT = "flushesQueuedCount";
   String MAX_FLUSH_QUEUE_SIZE = "maxFlushQueueSize";
+  String ROW_SEQUENCING_YIELDS = "rowSequencingYields";
   String COMPACTIONS_COMPLETED_DESC = "Number of compactions that have completed.";
   String COMPACTIONS_FAILED_DESC = "Number of compactions that have failed.";
   String LAST_MAJOR_COMPACTION_DESC = "Age of the last major compaction in milliseconds.";
@@ -57,6 +58,7 @@ public interface MetricsRegionSource extends Comparable<MetricsRegionSource> {
   String ROW_READS_ONLY_ON_MEMSTORE_DESC = "Row reads happening completely out of memstore";
   String MIXED_ROW_READS = "mixedRowReadsCount";
   String MIXED_ROW_READS_ON_STORE_DESC = "Row reads happening out of files and memstore on store";
+  String ROW_SEQUENCING_YIELDS_DESC = "Number of yields taken to sequence row commits";
 
   /**
    * Close the region's metrics as this region is closing.
@@ -99,5 +101,9 @@ public interface MetricsRegionSource extends Comparable<MetricsRegionSource> {
    */
   MetricsRegionAggregateSource getAggregateSource();
 
+  /**
+   * Update count of row sequencing yields.
+   */
+  void updateRowSequencingYields();
 
 }

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionSourceImpl.java

@@ -57,7 +57,7 @@ public class MetricsRegionSourceImpl implements MetricsRegionSource {
   private final String regionIncrementKey;
   private final String regionAppendKey;
   private final String regionScanKey;
-
+  private final String regionRowSequencerYieldsKey;
   /*
    * Implementation note: Do not put histograms per region. With hundreds of regions in a server
    * histograms allocate too many counters. See HBASE-17016.
@@ -69,6 +69,8 @@ public class MetricsRegionSourceImpl implements MetricsRegionSource {
   private final MutableFastCounter regionGet;
   private final MutableFastCounter regionScan;
 
+  private final MutableFastCounter regionRowSequencerYields;
+
   private final int hashCode;
 
   public MetricsRegionSourceImpl(MetricsRegionWrapper regionWrapper,
@@ -107,6 +109,10 @@ public MetricsRegionSourceImpl(MetricsRegionWrapper regionWrapper,
 
     regionScanKey = regionNamePrefix + MetricsRegionServerSource.SCAN_KEY + suffix;
     regionScan = registry.getCounter(regionScanKey, 0L);
+
+    regionRowSequencerYieldsKey = regionNamePrefix + MetricsRegionSource.ROW_SEQUENCING_YIELDS +
+        suffix;
+    regionRowSequencerYields = registry.getCounter(regionRowSequencerYieldsKey, 0L);
   }
 
   @Override
@@ -135,6 +141,7 @@ public void close() {
       registry.removeMetric(regionAppendKey);
       registry.removeMetric(regionGetKey);
       registry.removeMetric(regionScanKey);
+      registry.removeMetric(regionRowSequencerYieldsKey);
 
       regionWrapper = null;
     }
@@ -170,6 +177,11 @@ public void updateAppend() {
     regionAppend.incr();
   }
 
+  @Override
+  public void updateRowSequencingYields() {
+    regionRowSequencerYields.incr();
+  }
+
   @Override
   public MetricsRegionAggregateSource getAggregateSource() {
     return agg;
@@ -302,6 +314,10 @@ void snapshot(MetricsRecordBuilder mrb, boolean ignored) {
               regionNamePrefix + MetricsRegionSource.MAX_FLUSH_QUEUE_SIZE,
               MetricsRegionSource.MAX_FLUSH_QUEUE_DESC),
           this.regionWrapper.getMaxFlushQueueSize());
+      mrb.addCounter(Interns.info(
+             regionNamePrefix + MetricsRegionSource.ROW_SEQUENCING_YIELDS,
+             MetricsRegionSource.ROW_SEQUENCING_YIELDS_DESC),
+          this.regionWrapper.getRowSequencingYields());
       addCounter(mrb, this.regionWrapper.getMemstoreOnlyRowReadsCount(),
         MetricsRegionSource.ROW_READS_ONLY_ON_MEMSTORE,
         MetricsRegionSource.ROW_READS_ONLY_ON_MEMSTORE_DESC);

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionWrapper.java

@@ -183,4 +183,8 @@ public interface MetricsRegionWrapper {
    */
   Map<String, Long> getMixedRowReadsCount();
 
+  /**
+   * @return the number of yields made for row sequencing
+   */
+  long getRowSequencingYields();
 }

hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/regionserver/TestMetricsRegionSourceImpl.java

@@ -233,5 +233,10 @@ public Map<String, Long> getMixedRowReadsCount() {
       map.put("info", 0L);
       return map;
     }
+
+    @Override
+    public long getRowSequencingYields() {
+      return 0;
+    }
   }
 }