HBASE-30327 Implement adaptive parallel seek in StoreScanner

hbase-common/src/main/resources/hbase-default.xml

@@ -1010,6 +1010,19 @@ possible configurations would overwhelm and obscure the important.
       Enables StoreFileScanner parallel-seeking in StoreScanner,
       a feature which can reduce response latency under special conditions.</description>
   </property>
+  <property>
+    <name>hbase.storescanner.adaptive.parallel.seek.enable</name>
+    <value>false</value>
+    <description>
+      Enables the adaptive parallel seek strategy in StoreScanner, which dynamically
+      switches between parallel and sequential scanner seek execution based on thread
+      pool availability. When enabled, the scanner checks for available capacity in
+      the RS_PARALLEL_SEEK thread pool before submitting seek tasks; if the pool is
+      saturated, it falls back to sequential seeking and opportunistically parallelizes
+      remaining scanners as capacity becomes available. This setting only takes effect
+      when hbase.storescanner.parallel.seek.enable is also set to true.
+    </description>
+  </property>
   <property>
     <name>hbase.storescanner.parallel.seek.threads</name>
     <value>10</value>

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanInfo.java

@@ -45,6 +45,7 @@ public class ScanInfo {
   private boolean usePread;
   private long cellsPerTimeoutCheck;
   private boolean parallelSeekEnabled;
+  private boolean adaptiveParallelSeekEnabled;
   private final long preadMaxBytes;
   private final boolean newVersionBehavior;
 
@@ -92,13 +93,14 @@ public ScanInfo(Configuration conf, byte[] family, int minVersions, int maxVersi
       conf.getLong(HConstants.TABLE_MAX_ROWSIZE_KEY, HConstants.TABLE_MAX_ROWSIZE_DEFAULT),
       conf.getBoolean("hbase.storescanner.use.pread", false), getCellsPerTimeoutCheck(conf),
       conf.getBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, false),
-      conf.getLong(StoreScanner.STORESCANNER_PREAD_MAX_BYTES, 4 * blockSize), newVersionBehavior);
+      conf.getLong(StoreScanner.STORESCANNER_PREAD_MAX_BYTES, 4 * blockSize), newVersionBehavior,
+      conf.getBoolean(StoreScanner.STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE, false));
   }
 
   private ScanInfo(byte[] family, int minVersions, int maxVersions, long ttl,
     KeepDeletedCells keepDeletedCells, long timeToPurgeDeletes, CellComparator comparator,
     long tableMaxRowSize, boolean usePread, long cellsPerTimeoutCheck, boolean parallelSeekEnabled,
-    long preadMaxBytes, boolean newVersionBehavior) {
+    long preadMaxBytes, boolean newVersionBehavior, boolean adaptiveParallelSeekEnabled) {
     this.family = family;
     this.minVersions = minVersions;
     this.maxVersions = maxVersions;
@@ -112,6 +114,7 @@ private ScanInfo(byte[] family, int minVersions, int maxVersions, long ttl,
     this.parallelSeekEnabled = parallelSeekEnabled;
     this.preadMaxBytes = preadMaxBytes;
     this.newVersionBehavior = newVersionBehavior;
+    this.adaptiveParallelSeekEnabled = adaptiveParallelSeekEnabled;
   }
 
   long getTableMaxRowSize() {
@@ -130,6 +133,10 @@ boolean isParallelSeekEnabled() {
     return this.parallelSeekEnabled;
   }
 
+  boolean isAdaptiveParallelSeekEnabled() {
+    return this.adaptiveParallelSeekEnabled;
+  }
+
   public byte[] getFamily() {
     return family;
   }
@@ -181,7 +188,7 @@ ScanInfo customize(int maxVersions, long ttl, KeepDeletedCells keepDeletedCells,
     long timeToPurgeDeletes) {
     return new ScanInfo(family, minVersions, maxVersions, ttl, keepDeletedCells, timeToPurgeDeletes,
       comparator, tableMaxRowSize, usePread, cellsPerTimeoutCheck, parallelSeekEnabled,
-      preadMaxBytes, newVersionBehavior);
+      preadMaxBytes, newVersionBehavior, adaptiveParallelSeekEnabled);
   }
 
   @Override
@@ -192,6 +199,7 @@ public String toString() {
       .append("tableMaxRowSize", tableMaxRowSize).append("usePread", usePread)
       .append("cellsPerTimeoutCheck", cellsPerTimeoutCheck)
       .append("parallelSeekEnabled", parallelSeekEnabled).append("preadMaxBytes", preadMaxBytes)
-      .append("newVersionBehavior", newVersionBehavior).toString();
+      .append("newVersionBehavior", newVersionBehavior)
+      .append("adaptiveParallelSeekEnabled", adaptiveParallelSeekEnabled).toString();
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java

@@ -27,6 +27,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.IntConsumer;
@@ -46,6 +47,7 @@
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.conf.ConfigKey;
 import org.apache.hadoop.hbase.executor.ExecutorService;
+import org.apache.hadoop.hbase.executor.ExecutorType;
 import org.apache.hadoop.hbase.filter.Filter;
 import org.apache.hadoop.hbase.ipc.RpcCall;
 import org.apache.hadoop.hbase.ipc.RpcServer;
@@ -95,6 +97,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
    * A flag that enables StoreFileScanner parallel-seeking
    */
   private boolean parallelSeekEnabled = false;
+  private boolean adaptiveParallelSeekEnabled = false;
   private ExecutorService executor;
   private final Scan scan;
   private final long oldestUnexpiredTS;
@@ -129,6 +132,8 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
   static final boolean LAZY_SEEK_ENABLED_BY_DEFAULT = true;
   public static final String STORESCANNER_PARALLEL_SEEK_ENABLE =
     "hbase.storescanner.parallel.seek.enable";
+  public static final String STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE =
+    "hbase.storescanner.adaptive.parallel.seek.enable";
 
   /** Used during unit testing to ensure that lazy seek does save seek ops */
   private static boolean lazySeekEnabledGlobally = LAZY_SEEK_ENABLED_BY_DEFAULT;
@@ -232,6 +237,8 @@ private StoreScanner(HStore store, Scan scan, ScanInfo scanInfo, int numColumns,
       RegionServerServices rsService = store.getHRegion().getRegionServerServices();
       if (rsService != null && scanInfo.isParallelSeekEnabled()) {
         this.parallelSeekEnabled = true;
+        this.adaptiveParallelSeekEnabled =
+          this.parallelSeekEnabled && scanInfo.isAdaptiveParallelSeekEnabled();
         this.executor = rsService.getExecutorService();
       }
     }
@@ -441,6 +448,8 @@ protected void seekScanners(List<? extends KeyValueScanner> scanners, ExtendedCe
             totalScannersSoughtBytes += PrivateCellUtil.estimatedSerializedSizeOf(c);
           }
         }
+      } else if (adaptiveParallelSeekEnabled) {
+        adaptiveParallelSeek(scanners, seekKey);
       } else {
         parallelSeek(scanners, seekKey);
       }
@@ -1268,6 +1277,135 @@ private void parallelSeek(final List<? extends KeyValueScanner> scanners, final
     }
   }
 
+  /**
+   * Returns the number of threads available for immediate execution in the parallel seek thread
+   * pool. Uses a conservative approach: only reports capacity when the task queue is empty AND
+   * active threads < pool size.
+   * @return number of threads available for immediate execution, or 0 if saturated
+   */
+  private int getAvailableParallelSeekCapacity() {
+    ThreadPoolExecutor pool = executor.getExecutorThreadPool(ExecutorType.RS_PARALLEL_SEEK);
+    if (!pool.getQueue().isEmpty()) {
+      return 0; // Conservative: any queued work means saturated
+    }
+    return Math.max(0, pool.getCorePoolSize() - pool.getActiveCount());
+  }
+
+  /**
+   * Seeks scanners using an adaptive strategy that switches between parallel and sequential
+   * execution based on thread pool availability.
+   * <p>
+   * When the parallel seek thread pool is saturated, falls back to sequential seeking. After each
+   * sequential seek, re-checks capacity and opportunistically submits remaining scanners for
+   * parallel execution when slots become available.
+   * <p>
+   * If an IOException occurs during an inline seek, we must wait for any already-submitted handlers
+   * to complete before propagating the error. This prevents the caller from closing scanners that
+   * are still being used by worker threads.
+   *
+   * @param scanners list of KeyValueScanners to seek
+   * @param kv       the key to seek to
+   * @throws IOException if any seek operation fails
+   */
+  private void adaptiveParallelSeek(final List<? extends KeyValueScanner> scanners,
+    final ExtendedCell kv) throws IOException {
+    if (scanners.isEmpty()) return;
+
+    int scannerCount = scanners.size();
+    // Pre-count StoreFileScanners to size the latch correctly
+    int storeFileScannerCount = 0;
+    for (KeyValueScanner scanner : scanners) {
+      if (scanner instanceof StoreFileScanner) {
+        storeFileScannerCount++;
+      }
+    }
+    CountDownLatch latch = new CountDownLatch(storeFileScannerCount);
+    List<ParallelSeekHandler> handlers = new ArrayList<>(storeFileScannerCount);
+    int index = 0;
+    IOException inlineSeekError = null;
+
+    while (index < scannerCount) {
+      int capacity = getAvailableParallelSeekCapacity();
+
+      if (capacity == 0) {
+        // Sequential fallback: process one scanner on calling thread
+        KeyValueScanner scanner = scanners.get(index);
+        try {
+          scanner.seek(kv);
+        } catch (IOException e) {
+          // Must wait for already-submitted handlers before propagating error
+          inlineSeekError = e;
+          if (scanner instanceof StoreFileScanner) {
+            latch.countDown();
+          }
+          index++;
+          break;
+        }
+        if (scanner instanceof StoreFileScanner) {
+          latch.countDown();
+        }
+        index++;
+      } else {
+        // Opportunistic parallel: submit batch up to available capacity
+        int batchEnd = Math.min(index + capacity, scannerCount);
+        for (int i = index; i < batchEnd; i++) {
+          KeyValueScanner scanner = scanners.get(i);
+          if (scanner instanceof StoreFileScanner) {
+            ParallelSeekHandler seekHandler =
+              new ParallelSeekHandler(scanner, kv, this.readPt, latch);
+            executor.submit(seekHandler);
+            handlers.add(seekHandler);
+          } else {
+            try {
+              scanner.seek(kv);
+            } catch (IOException e) {
+              // Must wait for already-submitted handlers before propagating error
+              inlineSeekError = e;
+              // Count down latch for remaining StoreFileScanners in this batch that won't be
+              // processed
+              for (int j = i + 1; j < batchEnd; j++) {
+                if (scanners.get(j) instanceof StoreFileScanner) {
+                  latch.countDown();
+                }
+              }
+              index = batchEnd;
+              break;
+            }
+          }
+        }
+        if (inlineSeekError != null) {
+          break;
+        }
+        index = batchEnd;
+      }
+    }
+
+    // Count down latch for any remaining unprocessed StoreFileScanners
+    for (int i = index; i < scannerCount; i++) {
+      if (scanners.get(i) instanceof StoreFileScanner) {
+        latch.countDown();
+      }
+    }
+
+    try {
+      latch.await();
+    } catch (InterruptedException ie) {
+      throw (InterruptedIOException) new InterruptedIOException().initCause(ie);
+    }
+
+    // Check for errors from parallel handlers first
+    for (ParallelSeekHandler handler : handlers) {
+      if (handler.getErr() != null) {
+        throw new IOException(handler.getErr());
+      }
+    }
+
+    // Propagate inline seek error after all handlers have completed
+    if (inlineSeekError != null) {
+      throw inlineSeekError;
+    }
+  }
+
   /**
    * Used in testing.
    * @return all scanners in no particular order