diff --git a/cpp/velox/operators/plannodes/RowVectorStream.cc b/cpp/velox/operators/plannodes/RowVectorStream.cc index 4baa1f84712..0afcebbd111 100644 --- a/cpp/velox/operators/plannodes/RowVectorStream.cc +++ b/cpp/velox/operators/plannodes/RowVectorStream.cc @@ -136,47 +136,50 @@ void ValueStreamDataSource::addSplit(std::shared_ptr 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 \ No newline at end of file +} // namespace gluten diff --git a/cpp/velox/tests/ValueStreamDynamicFilterTest.cc b/cpp/velox/tests/ValueStreamDynamicFilterTest.cc index f9ba13bce6d..5c1f1561408 100644 --- a/cpp/velox/tests/ValueStreamDynamicFilterTest.cc +++ b/cpp/velox/tests/ValueStreamDynamicFilterTest.cc @@ -116,6 +116,74 @@ TEST_F(ValueStreamDynamicFilterTest, noFilterPassesAllRows) { ASSERT_EQ(ids, (std::vector{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 batches; + batches.reserve(kFilteredBatches + 2); + batches.push_back(makeRowVector({"id"}, {makeFlatVector({1, 2, 3})})); + for (int32_t i = 0; i < kFilteredBatches; i++) { + batches.push_back(makeRowVector({"id"}, {makeFlatVector({100, 200, 300})})); + } + batches.push_back(makeRowVector({"id"}, {makeFlatVector({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(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(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({1, 2, 3, 4, 5})});