HashJoinEngine: Check for interruptions while walking left cursor. (#16773)

* HashJoinEngine: Check for interruptions while walking left cursor.

Previously, the engine only checked for interruptions between emitting
joined rows. In scenarios where large numbers of left rows are skipped
completely (such as a highly selective INNER JOIN) this led to the
join cursor being insufficiently responsive to cancellation.

* Coverage.

processing/src/main/java/org/apache/druid/segment/join/HashJoinEngine.java

@@ -177,7 +177,7 @@ public class HashJoinEngine
       @Override
       public void advance()
       {
-        advanceUninterruptibly();
+        advance(true);
         BaseQuery.checkInterrupted();
       }
 
@@ -196,6 +196,11 @@ public class HashJoinEngine
 
       @Override
       public void advanceUninterruptibly()
+      {
+        advance(false);
+      }
+
+      private void advance(boolean interruptibly)
       {
         joinColumnSelectorFactory.advanceRowId();
 
@@ -217,7 +222,11 @@ public class HashJoinEngine
 
         do {
           // No more right-hand side matches; advance the left-hand side.
-          leftCursor.advanceUninterruptibly();
+          if (interruptibly) {
+            leftCursor.advance();
+          } else {
+            leftCursor.advanceUninterruptibly();
+          }
 
           // Update joinMatcher state to match new cursor position.
           matchCurrentPosition();

processing/src/test/java/org/apache/druid/segment/join/JoinTestHelper.java

@@ -342,6 +342,7 @@ public class JoinTestHelper
               .collect(Collectors.toList());
 
           final List<Object[]> rows = new ArrayList<>();
+          boolean interruptible = false; // test both advance() and advanceUninterruptibly()
 
           while (!cursor.isDone()) {
             final Object[] row = new Object[columns.size()];
@@ -351,7 +352,13 @@ public class JoinTestHelper
             }
 
             rows.add(row);
-            cursor.advance();
+            if (interruptible) {
+              cursor.advance();
+            } else {
+              cursor.advanceUninterruptibly();
+            }
+
+            interruptible = !interruptible;
           }
 
           return Sequences.simple(rows);