This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 2c32a92af7 [VL] Make ValueStreamDataSource::next() iterative instead
of recursive (#13346)
2c32a92af7 is described below
commit 2c32a92af7908c1a2f27560f53f2c1ce52508f95
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Tue Oct 20 12:26:42 2026 +0300
[VL] Make ValueStreamDataSource::next() iterative instead of recursive
(#13346)
---
cpp/velox/operators/plannodes/RowVectorStream.cc | 75 ++++++++++++------------
cpp/velox/tests/ValueStreamDynamicFilterTest.cc | 68 +++++++++++++++++++++
2 files changed, 107 insertions(+), 36 deletions(-)
diff --git a/cpp/velox/operators/plannodes/RowVectorStream.cc
b/cpp/velox/operators/plannodes/RowVectorStream.cc
index 4baa1f8471..4119eff037 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<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.
+ if (!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 f9ba13bce6..5c1f156140 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<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})});
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
