[VL] Make ValueStreamDataSource::next() iterative instead of recursive

cpp/velox/operators/plannodes/RowVectorStream.cc

@@ -136,47 +136,50 @@ void ValueStreamDataSource::addSplit(std::shared_ptr<facebook::velox::connector:
 std::optional<facebook::velox::RowVectorPtr> ValueStreamDataSource::next(
     uint64_t size,
     facebook::velox::ContinueFuture& future) {
-  // Try to get current iterator if we don't have one
-  while (!currentIterator_) {
-    if (pendingIterators_.empty()) {
-      // No more iterators to process
-      return nullptr;
-    }
-
-    // Get next RowVectorStream from queue
-    currentIterator_ = pendingIterators_.front();
-    pendingIterators_.erase(pendingIterators_.begin());
-  }
-
-  // Check if current stream has more data
-  if (!currentIterator_->hasNext()) {
-    // Current stream exhausted, try next one
-    currentIterator_ = nullptr;
-    return next(size, future); // Recursively try next stream
-  }
-
-  // Get next batch from current stream (RowVectorStream handles conversion)
-  auto rowVector = currentIterator_->next();
+  for (;;) {
+    // Try to get current iterator if we don't have one.
+    while (!currentIterator_) {
+      if (pendingIterators_.empty()) {
+        // No more iterators to process. Return an engaged null RowVectorPtr to
+        // tell TableScan the current split is finished, not connector-blocked.
+        return facebook::velox::RowVectorPtr(nullptr);
+      }
 
-  if (!rowVector) {
-    currentIterator_ = nullptr;
-    return next(size, future); // Recursively try next stream
-  }
+      // Get next RowVectorStream from queue.
+      currentIterator_ = pendingIterators_.front();
+      pendingIterators_.erase(pendingIterators_.begin());
+    }
 
-  // Update metrics
-  completedRows_ += rowVector->size();
-  completedBytes_ += rowVector->estimateFlatSize();
+    // Check if current stream has more data.
+    if (!currentIterator_->hasNext()) {
+      currentIterator_ = nullptr;
+      continue;
+    }
 
-  // Apply dynamic filters if any have been pushed down.
-  if (!dynamicFilters_.empty()) {
-    rowVector = applyDynamicFilters(rowVector);
+    // Get next batch from current stream (RowVectorStream handles conversion).
+    auto rowVector = currentIterator_->next();
     if (!rowVector) {
-      // All rows filtered out, try next batch.
-      return next(size, future);
+      // Current stream is done, move on to the next one.
+      currentIterator_ = nullptr;
+      continue;
     }
-  }
 
-  return rowVector;
+    // Update metrics.
+    completedRows_ += rowVector->size();
+    completedBytes_ += rowVector->estimateFlatSize();
+
+    // Apply dynamic filters if any have been pushed down.
+    if (!dynamicFilters_.empty()) {
+      rowVector = applyDynamicFilters(rowVector);
+      if (!rowVector) {
+        // All rows filtered out. Retry in-loop rather than recursing: a long run
+        // of fully-filtered batches must not grow the native stack.
+        continue;
+      }
+    }
+
+    return rowVector;
+  }
 }
 
 facebook::velox::RowVectorPtr ValueStreamDataSource::applyDynamicFilters(const facebook::velox::RowVectorPtr& input) {
@@ -278,4 +281,4 @@ void ValueStreamDataSource::applyFilterOnColumn(
   rows.updateBounds();
 }
 
-} // namespace gluten
+} // namespace gluten

cpp/velox/tests/ValueStreamDynamicFilterTest.cc

@@ -116,6 +116,74 @@ TEST_F(ValueStreamDynamicFilterTest, noFilterPassesAllRows) {
   ASSERT_EQ(ids, (std::vector<int64_t>{10, 20, 30}));
 }
 
+TEST_F(ValueStreamDynamicFilterTest, emptySplitProducesNoRows) {
+  auto outputType = ROW({"id"}, {BIGINT()});
+  auto scanNode = makeTableScanNode("vs-empty-split", outputType);
+
+  auto queryCtx = core::QueryCtx::create();
+  auto task = Task::create("test-empty-split", core::PlanFragment{scanNode}, 0, queryCtx, Task::ExecutionMode::kSerial);
+
+  task->addSplit(scanNode->id(), Split{makeSplit({})});
+  task->noMoreSplits(scanNode->id());
+
+  auto ids = readAllInt64(task.get());
+  ASSERT_TRUE(ids.empty());
+}
+
+// A long run of fully-filtered batches must be retried iteratively. The previous
+// implementation recursed once per eliminated batch, so this pattern grew the
+// native stack in proportion to the number of consecutive empty results.
+TEST_F(ValueStreamDynamicFilterTest, manyConsecutiveFullyFilteredBatches) {
+  constexpr int32_t kFilteredBatches = 4096;
+
+  // First batch passes the filter, every following batch is eliminated by it,
+  // and the last batch passes again.
+  std::vector<RowVectorPtr> batches;
+  batches.reserve(kFilteredBatches + 2);
+  batches.push_back(makeRowVector({"id"}, {makeFlatVector<int64_t>({1, 2, 3})}));
+  for (int32_t i = 0; i < kFilteredBatches; i++) {
+    batches.push_back(makeRowVector({"id"}, {makeFlatVector<int64_t>({100, 200, 300})}));
+  }
+  batches.push_back(makeRowVector({"id"}, {makeFlatVector<int64_t>({2})}));
+
+  auto outputType = asRowType(batches[0]->type());
+  auto scanNode = makeTableScanNode("vs-deep-filter", outputType);
+
+  auto queryCtx = core::QueryCtx::create();
+  auto task = Task::create("test-deep-filter", core::PlanFragment{scanNode}, 0, queryCtx, Task::ExecutionMode::kSerial);
+
+  task->addSplit(scanNode->id(), Split{makeSplit(std::move(batches))});
+  task->noMoreSplits(scanNode->id());
+
+  // First next() creates drivers and returns the first batch unfiltered.
+  ContinueFuture future = ContinueFuture::makeEmpty();
+  auto firstBatch = task->next(&future);
+  ASSERT_NE(firstBatch, nullptr);
+  ASSERT_EQ(firstBatch->size(), 3);
+
+  // Keep only id <= 3, which eliminates every one of the middle batches.
+  task->testingVisitDrivers([&](Driver* driver) {
+    auto* op = driver->findOperator(scanNode->id());
+    if (!op) {
+      return;
+    }
+    ASSERT_TRUE(op->canAddDynamicFilter());
+    PushdownFilters pf;
+    pf.filters[0] = std::make_shared<BigintRange>(1, 3, false);
+    pf.dynamicFilteredColumns.insert(0);
+    op->addDynamicFilterLocked("producer", pf);
+  });
+
+  // Skipping kFilteredBatches eliminated batches must not recurse.
+  auto lastBatch = task->next(&future);
+  ASSERT_NE(lastBatch, nullptr);
+  DecodedVector decoded(*lastBatch->childAt(0));
+  ASSERT_EQ(lastBatch->size(), 1);
+  ASSERT_EQ(decoded.valueAt<int64_t>(0), 2);
+
+  ASSERT_EQ(task->next(&future), nullptr);
+}
+
 // Test that filtering works when filter is injected after first batch.
 TEST_F(ValueStreamDynamicFilterTest, filterBigintRange) {
   auto batch1 = makeRowVector({"id"}, {makeFlatVector<int64_t>({1, 2, 3, 4, 5})});