Fix live update of Analysis tree for join with one index scan (#2473)

So far, the `Analysis` tree is updated live for join operations when both child operations produce their output lazily. However, when one or both child operations are index scans, no live updates are shown for them. Here is a simple example query that demonstrates both cases:
```sparql
SELECT * WHERE {
  ?s a ?o1, ?o2, ?o3
}
```
This change fixes the issue for the case where one child is an index scan and the other child produces its output lazily.

Co-authored-by: Robin Textor-Falconi <[email protected]>

Author: 2 people

src/engine/IndexScan.cpp

@@ -13,8 +13,6 @@
 #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"
 
@@ -366,7 +364,7 @@ Permutation::ScanSpecAndBlocks IndexScan::getScanSpecAndBlocks() const {
 }
 
 // _____________________________________________________________________________
-Permutation::IdTableGenerator IndexScan::getLazyScan(
+CompressedRelationReader::IdTableGeneratorInputRange IndexScan::getLazyScan(
     std::optional<std::vector<CompressedBlockMetadata>> blocks) const {
   // If there is a LIMIT or OFFSET clause that constrains the scan
   // (which can happen with an explicit subquery), we cannot use the prefiltered
@@ -378,12 +376,12 @@ Permutation::IdTableGenerator IndexScan::getLazyScan(
       scanSpecAndBlocks_, filteredBlocks, additionalColumns(),
       cancellationHandle_, locatedTriplesSnapshot(), getLimitOffset());
 
-  return cppcoro::fromInputRange(
-      ad_utility::InputRangeTypeErased<IdTable, LazyScanMetadata>(
-          ad_utility::CachingTransformInputRange<
-              ad_utility::OwningView<Permutation::IdTableGenerator>,
-              decltype(makeApplyColumnSubset()), LazyScanMetadata>{
-              std::move(lazyScanAllCols), makeApplyColumnSubset()}));
+  return CompressedRelationReader::IdTableGeneratorInputRange{
+      ad_utility::CachingTransformInputRange<
+          ad_utility::OwningView<
+              CompressedRelationReader::IdTableGeneratorInputRange>,
+          decltype(makeApplyColumnSubset()), LazyScanMetadata>{
+          std::move(lazyScanAllCols), makeApplyColumnSubset()}};
 };
 
 // _____________________________________________________________________________
@@ -394,7 +392,7 @@ std::optional<Permutation::MetadataAndBlocks> IndexScan::getMetadataForScan()
 };
 
 // _____________________________________________________________________________
-std::array<Permutation::IdTableGenerator, 2>
+std::array<CompressedRelationReader::IdTableGeneratorInputRange, 2>
 IndexScan::lazyScanForJoinOfTwoScans(const IndexScan& s1, const IndexScan& s2) {
   AD_CONTRACT_CHECK(s1.numVariables_ <= 3 && s2.numVariables_ <= 3);
   AD_CONTRACT_CHECK(s1.numVariables_ >= 1 && s2.numVariables_ >= 1);
@@ -442,7 +440,8 @@ IndexScan::lazyScanForJoinOfTwoScans(const IndexScan& s1, const IndexScan& s2) {
 }
 
 // _____________________________________________________________________________
-Permutation::IdTableGenerator IndexScan::lazyScanForJoinOfColumnWithScan(
+CompressedRelationReader::IdTableGeneratorInputRange
+IndexScan::lazyScanForJoinOfColumnWithScan(
     ql::span<const Id> joinColumn) const {
   AD_EXPENSIVE_CHECK(ql::ranges::is_sorted(joinColumn));
   AD_CORRECTNESS_CHECK(numVariables_ <= 3 && numVariables_ > 0);
@@ -461,9 +460,8 @@ Permutation::IdTableGenerator IndexScan::lazyScanForJoinOfColumnWithScan(
 
 // _____________________________________________________________________________
 void IndexScan::updateRuntimeInfoForLazyScan(const LazyScanMetadata& metadata) {
-  updateRuntimeInformationWhenOptimizedOut(
-      RuntimeInformation::Status::lazilyMaterialized);
   auto& rti = runtimeInfo();
+  rti.status_ = RuntimeInformation::Status::lazilyMaterialized;
   rti.numRows_ = metadata.numElementsYielded_;
   rti.totalTime_ = metadata.blockingTime_;
   rti.addDetail("num-blocks-read", metadata.numBlocksRead_);
@@ -481,6 +479,7 @@ void IndexScan::updateRuntimeInfoForLazyScan(const LazyScanMetadata& metadata) {
   updateIfPositive(metadata.numBlocksPostprocessed_,
                    "num-blocks-postprocessed");
   updateIfPositive(metadata.numBlocksWithUpdate_, "num-blocks-with-update");
+  signalQueryUpdate();
 }
 
 // Store a Generator and its corresponding iterator as well as unconsumed values
@@ -629,13 +628,24 @@ Result::LazyResult IndexScan::createPrefilteredJoinSide(
 Result::LazyResult IndexScan::createPrefilteredIndexScanSide(
     std::shared_ptr<SharedGeneratorState> innerState) {
   using LoopControl = ad_utility::LoopControl<Result::IdTableVocabPair>;
+  using namespace std::chrono_literals;
 
   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 scan = std::make_shared<
+              CompressedRelationReader::IdTableGeneratorInputRange>(
+              getLazyScan());
+          scan->details().numBlocksAll_ =
+              getMetadataForScan().value().sizeBlockMetadata_;
+          return LoopControl::breakWithYieldAll(
+              ad_utility::CachingTransformInputRange(*scan, [this, scan](
+                                                                auto& table) {
+                updateRuntimeInfoForLazyScan(scan->details());
+                return Result::IdTableVocabPair{std::move(table), LocalVocab{}};
+              }));
         }
 
         auto& pendingBlocks = state->pendingBlocks_;
@@ -648,6 +658,8 @@ Result::LazyResult IndexScan::createPrefilteredIndexScanSide(
           }
           state->fetch();
         }
+        metadata.numBlocksAll_ = state->metaBlocks_.sizeBlockMetadata_;
+        updateRuntimeInfoForLazyScan(metadata);
 
         // We now have non-empty pending blocks
         auto scan = getLazyScan(std::move(pendingBlocks));
@@ -658,19 +670,16 @@ Result::LazyResult IndexScan::createPrefilteredIndexScanSide(
 
         // 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 {
+            std::move(scan),
+            [&metadata, &scanDetails,
+             originalMetadata = metadata](auto& table) mutable {
+              // Make sure we don't add everything more than once.
+              metadata = originalMetadata;
               metadata.aggregate(scanDetails);
+              return Result::IdTableVocabPair{std::move(table), LocalVocab{}};
             });
 
-        auto scanWithCallback = ad_utility::CallbackOnEndView{
-            std::move(transformedScan), std::move(callback)};
-        return LoopControl::yieldAll(std::move(scanWithCallback));
+        return LoopControl::yieldAll(std::move(transformedScan));
       }};
   return Result::LazyResult{std::move(range)};
 }

src/engine/IndexScan.h

@@ -97,15 +97,15 @@ class IndexScan final : public Operation {
   // blocks, but only the blocks that can theoretically contain matching rows
   // when performing a join on the first column of the result of `s1` with the
   // first column of the result of `s2`.
-  static std::array<Permutation::IdTableGenerator, 2> lazyScanForJoinOfTwoScans(
-      const IndexScan& s1, const IndexScan& s2);
+  static std::array<CompressedRelationReader::IdTableGeneratorInputRange, 2>
+  lazyScanForJoinOfTwoScans(const IndexScan& s1, const IndexScan& s2);
 
   // Return a generator that lazily yields the result in blocks, but only
   // the blocks that can theoretically contain matching rows when performing a
   // join between the first column of the result with the `joinColumn`.
   // Requires that the `joinColumn` is sorted, else the behavior is undefined.
-  Permutation::IdTableGenerator lazyScanForJoinOfColumnWithScan(
-      ql::span<const Id> joinColumn) const;
+  CompressedRelationReader::IdTableGeneratorInputRange
+  lazyScanForJoinOfColumnWithScan(ql::span<const Id> joinColumn) const;
 
   // Return two generators, the first of which yields exactly the elements of
   // `input` and the second of which yields the matching blocks, skipping the
@@ -237,7 +237,7 @@ class IndexScan final : public Operation {
 
   // Helper functions for the public `getLazyScanFor...` methods and
   // `chunkedIndexScan` (see above).
-  Permutation::IdTableGenerator getLazyScan(
+  CompressedRelationReader::IdTableGeneratorInputRange getLazyScan(
       std::optional<std::vector<CompressedBlockMetadata>> blocks =
           std::nullopt) const;
   std::optional<Permutation::MetadataAndBlocks> getMetadataForScan() const;

src/engine/JoinHelpers.h

@@ -92,10 +92,11 @@ using GeneratorWithDetails =
 // Convert a `generator<IdTable` to a `generator<IdTableAndFirstCol>` for more
 // efficient access in the join columns below.
 inline GeneratorWithDetails convertGenerator(
-    Permutation::IdTableGenerator gen) {
+    CompressedRelationReader::IdTableGeneratorInputRange gen) {
   // Store the generator in a wrapper so we can access its details after moving
   auto generatorStorage =
-      std::make_shared<Permutation::IdTableGenerator>(std::move(gen));
+      std::make_shared<CompressedRelationReader::IdTableGeneratorInputRange>(
+          std::move(gen));
 
   // Create the range with a pointer to the generator's details
   auto range = InputRangeTypeErased<IdTableAndFirstCol<IdTable>>(

src/index/Permutation.cpp

@@ -8,7 +8,6 @@
 
 #include "index/ConstantsIndexBuilding.h"
 #include "index/DeltaTriples.h"
-#include "util/GeneratorConverter.h"
 #include "util/StringUtils.h"
 
 // _____________________________________________________________________
@@ -194,7 +193,7 @@ std::optional<Permutation::MetadataAndBlocks> Permutation::getMetadataAndBlocks(
 }
 
 // _____________________________________________________________________
-Permutation::IdTableGenerator Permutation::lazyScan(
+CompressedRelationReader::IdTableGeneratorInputRange Permutation::lazyScan(
     const ScanSpecAndBlocks& scanSpecAndBlocks,
     std::optional<std::vector<CompressedBlockMetadata>> optBlocks,
     ColumnIndicesRef additionalColumns,
@@ -207,14 +206,10 @@ Permutation::IdTableGenerator Permutation::lazyScan(
     optBlocks = CompressedRelationReader::convertBlockMetadataRangesToVector(
         scanSpecAndBlocks.blockMetadata_);
   }
-  auto lazyScan{p.reader().lazyScan(
+  return p.reader().lazyScan(
       scanSpecAndBlocks.scanSpec_, std::move(optBlocks.value()),
       std::move(columns), cancellationHandle,
-      p.getLocatedTriplesForPermutation(locatedTriplesSnapshot), limitOffset)};
-
-  return cppcoro::fromInputRange<IdTable,
-                                 CompressedRelationReader::LazyScanMetadata>(
-      std::move(lazyScan));
+      p.getLocatedTriplesForPermutation(locatedTriplesSnapshot), limitOffset);
 }
 
 // ______________________________________________________________________

src/index/Permutation.h

@@ -89,8 +89,6 @@ class Permutation {
   using MetadataAndBlocks =
       CompressedRelationReader::ScanSpecAndBlocksAndBounds;
 
-  using IdTableGenerator = CompressedRelationReader::IdTableGenerator;
-
   // The function `lazyScan` is similar to `scan` (see above) with
   // the following differences:
   // - The result is returned as a lazy generator of blocks.
@@ -108,7 +106,7 @@ class Permutation {
   // TODO<joka921> We should only communicate this interface via the
   // `ScanSpecAndBlocksAndBounds` class and make this a strong class that always
   // maintains its invariants.
-  IdTableGenerator lazyScan(
+  CompressedRelationReader::IdTableGeneratorInputRange lazyScan(
       const ScanSpecAndBlocks& scanSpecAndBlocks,
       std::optional<std::vector<CompressedBlockMetadata>> optBlocks,
       ColumnIndicesRef additionalColumns,

src/util/GeneratorConverter.h

@@ -14,7 +14,6 @@
 #include "backports/three_way_comparison.h"
 #include "util/AsyncStream.h"
 #include "util/CompressorStream.h"
-#include "util/GeneratorConverter.h"
 #include "util/StringUtils.h"
 #include "util/TypeTraits.h"
 #include "util/http/MediaTypes.h"
@@ -168,7 +167,11 @@ CPP_template(typename RequestType)(
       ad_utility::content_encoding::getCompressionMethodForRequest(request);
 
   auto asyncGenerator = streams::runStreamAsync(std::move(generator), 100);
-  auto coroAsyncGenerator = cppcoro::fromInputRange(std::move(asyncGenerator));
+  auto coroAsyncGenerator = [](auto range) -> cppcoro::generator<std::string> {
+    for (auto& value : range) {
+      co_yield value;
+    }
+  }(std::move(asyncGenerator));
 
   if (method != CompressionMethod::NONE) {
     response.body() =