[native] Advance velox.

prestodb · Jan 25, 2025 · 13cb729 · 13cb729
1 parent f3342e8
commit 13cb729
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 6 deletions.
diff --git a/presto-native-execution/presto_cpp/main/tests/PrestoExchangeSourceTest.cpp b/presto-native-execution/presto_cpp/main/tests/PrestoExchangeSourceTest.cpp
@@ -424,12 +424,14 @@ std::unique_ptr<exec::SerializedPage> waitForNextPage(
     const std::shared_ptr<exec::ExchangeQueue>& queue) {
   bool atEnd;
   facebook::velox::ContinueFuture future;
-  auto pages = queue->dequeueLocked(1, &atEnd, &future);
+  ContinuePromise stalePromise = ContinuePromise::makeEmpty();
+  auto pages = queue->dequeueLocked(0, 1, &atEnd, &future, &stalePromise);
   EXPECT_LE(pages.size(), 1);
   EXPECT_FALSE(atEnd);
   if (pages.empty()) {
     std::move(future).get();
-    pages = queue->dequeueLocked(1, &atEnd, &future);
+    ContinuePromise stalePromise = ContinuePromise::makeEmpty();
+    pages = queue->dequeueLocked(0, 1, &atEnd, &future, &stalePromise);
     EXPECT_EQ(pages.size(), 1);
   }
   return std::move(pages.front());
@@ -438,11 +440,13 @@ std::unique_ptr<exec::SerializedPage> waitForNextPage(
 void waitForEndMarker(const std::shared_ptr<exec::ExchangeQueue>& queue) {
   bool atEnd;
   facebook::velox::ContinueFuture future;
-  auto pages = queue->dequeueLocked(1, &atEnd, &future);
+  ContinuePromise stalePromise = ContinuePromise::makeEmpty();
+  auto pages = queue->dequeueLocked(0, 1, &atEnd, &future, &stalePromise);
   ASSERT_TRUE(pages.empty());
   if (!atEnd) {
     std::move(future).get();
-    pages = queue->dequeueLocked(1, &atEnd, &future);
+    ContinuePromise stalePromise = ContinuePromise::makeEmpty();
+    pages = queue->dequeueLocked(0, 1, &atEnd, &future, &stalePromise);
     ASSERT_TRUE(pages.empty());
     ASSERT_TRUE(atEnd);
   }
@@ -525,7 +529,7 @@ class PrestoExchangeSourceTest : public ::testing::TestWithParam<Params> {
   }
 
   std::shared_ptr<exec::ExchangeQueue> makeSingleSourceQueue() {
-    auto queue = std::make_shared<exec::ExchangeQueue>();
+    auto queue = std::make_shared<exec::ExchangeQueue>(1, 0);
     queue->addSourceLocked();
     queue->noMoreSources();
     return queue;

diff --git a/presto-native-execution/velox b/presto-native-execution/velox