Disable the driver suspension optimization for table scan (#9569)

Summary:
In Meta internal shadow testing, we found driver suspension in table scan get output could
cause high driver queue latency which slow down the query as the suspension leave does
on-thread sleep. We might need to extend it to support yield from leave if driver is pauses so
that we do off-thread sleep. Also the suspension leave needs to take care of task terminate
and throws after that, otherwise the driver might proceed to the next op which might cause
random failure if the task has been terminated as discovered by join fuzzer.
As for now, we disable this optimization and re-enable this after we have the fixes for the above
two issues later.

Pull Request resolved: #9569

Reviewed By: kevinwilfong, oerling

Differential Revision: D56449266

Pulled By: xiaoxmeng

fbshipit-source-id: c60c620e0a45ad1f4f878b02c02b0327ab101faf

velox/exec/TableScan.cpp

@@ -61,15 +61,10 @@ bool TableScan::shouldYield(StopReason taskStopReason, size_t startTimeMs)
     const {
   // Checks task-level yield signal, driver-level yield signal and table scan
   // output processing time limit.
-  //
-  // NOTE: if the task is being paused, then we shall continue execution as we
-  // won't yield the driver thread but simply spinning (with on-thread time
-  // sleep) until the task has been resumed.
-  return (taskStopReason == StopReason::kYield ||
-          driverCtx_->driver->shouldYield() ||
-          ((getOutputTimeLimitMs_ != 0) &&
-           (getCurrentTimeMs() - startTimeMs) >= getOutputTimeLimitMs_)) &&
-      !driverCtx_->task->pauseRequested();
+  return taskStopReason == StopReason::kYield ||
+      driverCtx_->driver->shouldYield() ||
+      ((getOutputTimeLimitMs_ != 0) &&
+       (getCurrentTimeMs() - startTimeMs) >= getOutputTimeLimitMs_);
 }
 
 bool TableScan::shouldStop(StopReason taskStopReason) const {
@@ -78,7 +73,6 @@ bool TableScan::shouldStop(StopReason taskStopReason) const {
 }
 
 RowVectorPtr TableScan::getOutput() {
-  SuspendedSection suspendedSection(driverCtx_->driver);
   auto exitCurStatusGuard = folly::makeGuard([this]() { curStatus_ = ""; });
 
   if (noMoreSplits_) {

velox/exec/tests/TableScanTest.cpp

@@ -4127,7 +4127,9 @@ TEST_F(TableScanTest, partitionKeyNotMatchPartitionKeysHandle) {
   assertQuery(op, split, "SELECT c0 FROM tmp");
 }
 
-TEST_F(TableScanTest, memoryArbitrationWithSlowTableScan) {
+// TODO: re-enable this test once we add back driver suspension support for
+// table scan.
+TEST_F(TableScanTest, DISABLED_memoryArbitrationWithSlowTableScan) {
   const size_t numFiles{2};
   std::vector<std::shared_ptr<TempFilePath>> filePaths;
   std::vector<RowVectorPtr> vectorsForDuckDb;
@@ -4205,7 +4207,11 @@ TEST_F(TableScanTest, memoryArbitrationWithSlowTableScan) {
   queryThread.join();
 }
 
-DEBUG_ONLY_TEST_F(TableScanTest, memoryArbitrationByTableScanAllocation) {
+// TODO: re-enable this test once we add back driver suspension support for
+// table scan.
+DEBUG_ONLY_TEST_F(
+    TableScanTest,
+    DISABLED_memoryArbitrationByTableScanAllocation) {
   auto vectors = makeVectors(10, 1'000);
   auto filePath = TempFilePath::create();
   writeToFile(filePath->getPath(), vectors);