Migrate coroutines in src/engine/IndexScan to C++17 (#2316)

Co-authored-by: Johannes Kalmbach <johannes.kalmbach@gmail.com>

Author: 2 people

src/engine/IndexScan.cpp

@@ -13,6 +13,10 @@
 #include "engine/QueryExecutionTree.h"
 #include "index/IndexImpl.h"
 #include "parser/ParsedQuery.h"
+#include "util/Generator.h"
+#include "util/GeneratorConverter.h"
+#include "util/InputRangeUtils.h"
+#include "util/Iterators.h"
 
 using std::string;
 using LazyScanMetadata = CompressedRelationReader::LazyScanMetadata;
@@ -235,10 +239,11 @@ IndexScan::makeCopyWithPrefilteredScanSpecAndBlocks(
 }
 
 // _____________________________________________________________________________
-Result::Generator IndexScan::chunkedIndexScan() const {
-  for (IdTable& idTable : getLazyScan()) {
-    co_yield {std::move(idTable), LocalVocab{}};
-  }
+Result::LazyResult IndexScan::chunkedIndexScan() const {
+  return Result::LazyResult{
+      ad_utility::CachingTransformInputRange(getLazyScan(), [](auto& table) {
+        return Result::IdTableVocabPair{std::move(table), LocalVocab{}};
+      })};
 }
 
 // _____________________________________________________________________________
@@ -372,13 +377,13 @@ Permutation::IdTableGenerator IndexScan::getLazyScan(
   auto lazyScanAllCols = getScanPermutation().lazyScan(
       scanSpecAndBlocks_, filteredBlocks, additionalColumns(),
       cancellationHandle_, locatedTriplesSnapshot(), getLimitOffset());
-  auto& detailsRef = co_await cppcoro::getDetails;
-  lazyScanAllCols.setDetailsPointer(&detailsRef);
-  auto applySubset = makeApplyColumnSubset();
 
-  for (auto& table : lazyScanAllCols) {
-    co_yield applySubset(std::move(table));
-  }
+  return cppcoro::fromInputRange(
+      ad_utility::InputRangeTypeErased<IdTable, LazyScanMetadata>(
+          ad_utility::CachingTransformInputRange<
+              ad_utility::OwningView<Permutation::IdTableGenerator>,
+              decltype(makeApplyColumnSubset()), LazyScanMetadata>{
+              std::move(lazyScanAllCols), makeApplyColumnSubset()}));
 };
 
 // _____________________________________________________________________________
@@ -514,10 +519,7 @@ struct IndexScan::SharedGeneratorState {
       iterator_ = generator_.begin();
     }
     auto& iterator = iterator_.value();
-    while (iterator != generator_.end()) {
-      if (!iterator->idTable_.empty()) {
-        break;
-      }
+    while (iterator != generator_.end() && iterator->idTable_.empty()) {
       ++iterator;
     }
     doneFetching_ = iterator_ == generator_.end();
@@ -586,75 +588,104 @@ struct IndexScan::SharedGeneratorState {
 };
 
 // _____________________________________________________________________________
-Result::Generator IndexScan::createPrefilteredJoinSide(
+Result::LazyResult IndexScan::createPrefilteredJoinSide(
     std::shared_ptr<SharedGeneratorState> innerState) {
-  if (innerState->hasUndef()) {
-    AD_CORRECTNESS_CHECK(innerState->prefetchedValues_.empty());
-    for (auto& value : ql::ranges::subrange{innerState->iterator_.value(),
-                                            innerState->generator_.end()}) {
-      co_yield value;
-    }
-    co_return;
-  }
-  auto& prefetchedValues = innerState->prefetchedValues_;
-  while (true) {
-    if (prefetchedValues.empty()) {
-      if (innerState->doneFetching_) {
-        co_return;
-      }
-      innerState->fetch();
-      AD_CORRECTNESS_CHECK(!prefetchedValues.empty() ||
-                           innerState->doneFetching_);
-    }
-    // Make a defensive copy of the values to avoid modification during
-    // iteration when yielding.
-    auto copy = std::move(prefetchedValues);
-    // Moving out does not necessarily clear the values, so we do it explicitly.
-    prefetchedValues.clear();
-    for (auto& value : copy) {
-      co_yield value;
-    }
-  }
+  using LoopControl = ad_utility::LoopControl<Result::IdTableVocabPair>;
+
+  auto range = ad_utility::InputRangeFromLoopControlGet{
+      [state = std::move(innerState)]() mutable {
+        // Handle UNDEF case: pass through remaining input
+        if (state->hasUndef()) {
+          if (!state->iterator_.has_value()) {
+            state->iterator_ = state->generator_.begin();
+          }
+          return LoopControl::breakWithYieldAll(ql::ranges::subrange(
+              state->iterator_.value(), state->generator_.end()));
+        }
+
+        auto& prefetched = state->prefetchedValues_;
+
+        if (prefetched.empty() && !state->doneFetching_) {
+          state->fetch();
+        }
+
+        if (prefetched.empty()) {
+          AD_CORRECTNESS_CHECK(state->doneFetching_);
+          return LoopControl::makeBreak();
+        }
+
+        // Make a defensive copy of the values to avoid modification during
+        // iteration when yielding.
+        auto copy = std::move(prefetched);
+        prefetched.clear();
+
+        // Yield all the newly found values
+        return LoopControl::yieldAll(std::move(copy));
+      }};
+  return Result::LazyResult{std::move(range)};
 }
 
 // _____________________________________________________________________________
-Result::Generator IndexScan::createPrefilteredIndexScanSide(
+Result::LazyResult IndexScan::createPrefilteredIndexScanSide(
     std::shared_ptr<SharedGeneratorState> innerState) {
-  if (innerState->hasUndef()) {
-    for (auto& pair : chunkedIndexScan()) {
-      co_yield pair;
-    }
-    co_return;
-  }
-  LazyScanMetadata metadata;
-  auto& pendingBlocks = innerState->pendingBlocks_;
-  while (true) {
-    if (pendingBlocks.empty()) {
-      if (innerState->doneFetching_) {
-        metadata.numBlocksAll_ = innerState->metaBlocks_.sizeBlockMetadata_;
-        updateRuntimeInfoForLazyScan(metadata);
-        co_return;
-      }
-      innerState->fetch();
-    }
-    auto scan = getLazyScan(std::move(pendingBlocks));
-    AD_CORRECTNESS_CHECK(pendingBlocks.empty());
-    for (IdTable& idTable : scan) {
-      co_yield {std::move(idTable), LocalVocab{}};
-    }
-    metadata.aggregate(scan.details());
-  }
+  using LoopControl = ad_utility::LoopControl<Result::IdTableVocabPair>;
+
+  auto range = ad_utility::InputRangeFromLoopControlGet{
+      [this, state = std::move(innerState),
+       metadata = LazyScanMetadata{}]() mutable {
+        // Handle UNDEF case using LoopControl pattern
+        if (state->hasUndef()) {
+          return LoopControl::breakWithYieldAll(chunkedIndexScan());
+        }
+
+        auto& pendingBlocks = state->pendingBlocks_;
+
+        while (pendingBlocks.empty()) {
+          if (state->doneFetching_) {
+            metadata.numBlocksAll_ = state->metaBlocks_.sizeBlockMetadata_;
+            updateRuntimeInfoForLazyScan(metadata);
+            return LoopControl::makeBreak();
+          }
+          state->fetch();
+        }
+
+        // We now have non-empty pending blocks
+        auto scan = getLazyScan(std::move(pendingBlocks));
+        AD_CORRECTNESS_CHECK(pendingBlocks.empty());
+
+        // Capture scan details by reference so we get the updated values
+        const auto& scanDetails = scan.details();
+
+        // Transform the scan to Result::IdTableVocabPair and yield all
+        auto transformedScan = ad_utility::CachingTransformInputRange(
+            std::move(scan), [](auto& table) {
+              return Result::IdTableVocabPair{std::move(table), LocalVocab{}};
+            });
+
+        // Use CallbackOnEndView to aggregate metadata after scan is consumed
+        auto callback = ad_utility::makeAssignableLambda(
+            [&metadata, &scanDetails]() mutable {
+              metadata.aggregate(scanDetails);
+            });
+
+        auto scanWithCallback = ad_utility::CallbackOnEndView{
+            std::move(transformedScan), std::move(callback)};
+        return LoopControl::yieldAll(std::move(scanWithCallback));
+      }};
+  return Result::LazyResult{std::move(range)};
 }
 
 // _____________________________________________________________________________
-std::pair<Result::Generator, Result::Generator> IndexScan::prefilterTables(
+std::pair<Result::LazyResult, Result::LazyResult> IndexScan::prefilterTables(
     Result::LazyResult input, ColumnIndex joinColumn) {
   AD_CORRECTNESS_CHECK(numVariables_ <= 3 && numVariables_ > 0);
   auto metaBlocks = getMetadataForScan();
 
   if (!metaBlocks.has_value()) {
-    return {Result::Generator{}, Result::Generator{}};
+    // Return empty results
+    return {Result::LazyResult{}, Result::LazyResult{}};
   }
+
   auto state = std::make_shared<SharedGeneratorState>(
       std::move(input), joinColumn, std::move(metaBlocks.value()));
   return {createPrefilteredJoinSide(state),

src/engine/IndexScan.h

@@ -113,24 +113,24 @@ class IndexScan final : public Operation {
   // `input` to speed up join algorithms when no undef values are presend. When
   // there are undef values, the second generator represents the full index
   // scan.
-  std::pair<Result::Generator, Result::Generator> prefilterTables(
+  std::pair<Result::LazyResult, Result::LazyResult> prefilterTables(
       Result::LazyResult input, ColumnIndex joinColumn);
 
  private:
-  // Implementation detail that allows to consume a generator from two other
-  // cooperating generators. Needs to be forward declared as it is used by
+  // Implementation detail that allows to consume a lazy range from two other
+  // cooperating ranges. Needs to be forward declared as it is used by
   // several member functions below.
   struct SharedGeneratorState;
 
-  // Helper function that creates a generator that re-yields the generator
+  // Helper function that creates a lazy range that re-yields the input
   // wrapped by `innerState`.
-  static Result::Generator createPrefilteredJoinSide(
+  static Result::LazyResult createPrefilteredJoinSide(
       std::shared_ptr<SharedGeneratorState> innerState);
 
-  // Helper function that creates a generator yielding prefiltered rows of this
-  // index scan according to the block metadata, that match the tables yielded
-  // by the generator wrapped by `innerState`.
-  Result::Generator createPrefilteredIndexScanSide(
+  // Helper function that creates a lazy range yielding prefiltered rows of
+  // this index scan according to the block metadata, that match the tables
+  // yielded by the input wrapped by `innerState`.
+  Result::LazyResult createPrefilteredIndexScanSide(
       std::shared_ptr<SharedGeneratorState> innerState);
 
   // TODO<joka921> Make the `getSizeEstimateBeforeLimit()` function `const` for
@@ -220,7 +220,7 @@ class IndexScan final : public Operation {
       ScanSpecAndBlocks scanSpecAndBlocks) const;
 
   // Return the (lazy) `IdTable` for this `IndexScan` in chunks.
-  Result::Generator chunkedIndexScan() const;
+  Result::LazyResult chunkedIndexScan() const;
   // Get the `IdTable` for this `IndexScan` in one piece.
   IdTable materializedIndexScan() const;
 

src/util/Generator.h

@@ -22,6 +22,15 @@ namespace cppcoro {
 struct GetDetails {};
 static constexpr GetDetails getDetails;
 
+template <typename Details>
+struct SetDetailsPointer {
+  Details* pointer_;
+};
+template <typename Details>
+struct SetDetails {
+  Details details_;
+};
+
 // This struct is used as the default of the details object for the case that
 // there are no details
 struct NoDetails {};
@@ -93,7 +102,40 @@ class generator_promise {
     return {*this};
   }
 
+  struct SetDetailsPointerAwaiter {
+    SetDetailsPointerAwaiter(generator_promise& promise,
+                             struct SetDetailsPointer<Details> details) {
+      promise.setDetailsPointer(details.pointer_);
+    }
+    constexpr bool await_ready() const { return true; }
+    constexpr bool await_suspend(std::coroutine_handle<>) const noexcept {
+      return false;
+    }
+    constexpr void await_resume() const noexcept {}
+  };
+
+  struct SetDetailsAwaiter {
+    SetDetailsAwaiter(generator_promise& promise,
+                      struct SetDetails<Details> details) {
+      promise.setDetails(std::move(details.details_));
+    }
+    constexpr bool await_ready() const { return true; }
+    constexpr bool await_suspend(std::coroutine_handle<>) const noexcept {
+      return false;
+    }
+    constexpr void await_resume() const noexcept {}
+  };
+
   static constexpr bool hasDetails = !std::is_same_v<Details, NoDetails>;
+  SetDetailsPointerAwaiter await_transform(SetDetailsPointer<Details> details)
+      requires hasDetails {
+    return {*this, details};
+  }
+  SetDetailsAwaiter await_transform(SetDetails<Details> details)
+      requires hasDetails {
+    return {*this, details};
+  }
+
   Details& details() requires hasDetails {
     return std::holds_alternative<Details>(m_details)
                ? std::get<Details>(m_details)
@@ -105,6 +147,10 @@ class generator_promise {
     m_details = pointer;
   }
 
+  void setDetails(Details details) requires hasDetails {
+    m_details = std::move(details);
+  }
+
  private:
   pointer_type m_value;
   std::exception_ptr m_exception;
@@ -295,7 +341,6 @@ T getSingleElement(generator<T, Details> g) {
   AD_CORRECTNESS_CHECK(++it == g.end());
   return t;
 }
-
 }  // namespace cppcoro
 
 #endif

src/util/GeneratorConverter.h

@@ -18,13 +18,18 @@ namespace cppcoro {
 template <typename T, typename D>
 generator<T, D> fromInputRange(ad_utility::InputRangeTypeErased<T, D> range) {
   if constexpr (!std::is_same_v<D, ad_utility::NoDetails>) {
-    auto& detailsRef = co_await cppcoro::getDetails;
-    range.setDetailsPointer(&detailsRef);
+    co_await cppcoro::SetDetailsPointer<D>{&range.details()};
   }
 
   for (auto& value : range) {
     co_yield value;
   }
+
+  // The range is about to be destroyed, copy the details.
+  if constexpr (!std::is_same_v<D, ad_utility::NoDetails>) {
+    co_await cppcoro::SetDetails<std::decay_t<decltype(range.details())>>{
+        std::move(range.details())};
+  }
 }
 }  // namespace cppcoro