Provide framework for generic lazily evaluated operation results

src/engine/Bind.cpp

@@ -81,7 +81,7 @@ std::vector<QueryExecutionTree*> Bind::getChildren() {
 }
 
 // _____________________________________________________________________________
-Result Bind::computeResult([[maybe_unused]] bool requestLaziness) {
+ProtoResult Bind::computeResult([[maybe_unused]] bool requestLaziness) {
   using std::endl;
   LOG(DEBUG) << "Get input to BIND operation..." << endl;
   std::shared_ptr<const Result> subRes = _subtree->getResult();

src/engine/Bind.h

@@ -46,7 +46,7 @@ class Bind : public Operation {
   [[nodiscard]] vector<ColumnIndex> resultSortedOn() const override;
 
  private:
-  Result computeResult([[maybe_unused]] bool requestLaziness) override;
+  ProtoResult computeResult([[maybe_unused]] bool requestLaziness) override;
 
   // Implementation for the binding of arbitrary expressions.
   template <size_t IN_WIDTH, size_t OUT_WIDTH>

src/engine/CartesianProductJoin.cpp

@@ -132,7 +132,7 @@ void CartesianProductJoin::writeResultColumn(std::span<Id> targetColumn,
   }
 }
 // ____________________________________________________________________________
-Result CartesianProductJoin::computeResult(
+ProtoResult CartesianProductJoin::computeResult(
     [[maybe_unused]] bool requestLaziness) {
   IdTable result{getExecutionContext()->getAllocator()};
   result.setNumColumns(getResultWidth());

src/engine/CartesianProductJoin.h

@@ -79,7 +79,7 @@ class CartesianProductJoin : public Operation {
 
  private:
   //! Compute the result of the query-subtree rooted at this element..
-  Result computeResult([[maybe_unused]] bool requestLaziness) override;
+  ProtoResult computeResult([[maybe_unused]] bool requestLaziness) override;
 
   // Copy each element from the `inputColumn` `groupSize` times to the
   // `targetColumn`. Repeat until the `targetColumn` is completely filled. Skip

src/engine/CountAvailablePredicates.cpp

@@ -100,7 +100,7 @@ size_t CountAvailablePredicates::getCostEstimate() {
 }
 
 // _____________________________________________________________________________
-Result CountAvailablePredicates::computeResult(
+ProtoResult CountAvailablePredicates::computeResult(
     [[maybe_unused]] bool requestLaziness) {
   LOG(DEBUG) << "CountAvailablePredicates result computation..." << std::endl;
   IdTable idTable{getExecutionContext()->getAllocator()};

src/engine/CountAvailablePredicates.h

@@ -103,6 +103,6 @@ class CountAvailablePredicates : public Operation {
   void computePatternTrickAllEntities(
       IdTable* result, const CompactVectorOfStrings<Id>& patterns) const;
 
-  Result computeResult([[maybe_unused]] bool requestLaziness) override;
+  ProtoResult computeResult([[maybe_unused]] bool requestLaziness) override;
   [[nodiscard]] VariableToColumnMap computeVariableToColumnMap() const override;
 };

src/engine/Distinct.cpp

@@ -38,7 +38,7 @@ VariableToColumnMap Distinct::computeVariableToColumnMap() const {
 }
 
 // _____________________________________________________________________________
-Result Distinct::computeResult([[maybe_unused]] bool requestLaziness) {
+ProtoResult Distinct::computeResult([[maybe_unused]] bool requestLaziness) {
   IdTable idTable{getExecutionContext()->getAllocator()};
   LOG(DEBUG) << "Getting sub-result for distinct result computation..." << endl;
   std::shared_ptr<const Result> subRes = _subtree->getResult();

src/engine/Distinct.h

@@ -55,7 +55,8 @@ class Distinct : public Operation {
   [[nodiscard]] string getCacheKeyImpl() const override;
 
  private:
-  virtual Result computeResult([[maybe_unused]] bool requestLaziness) override;
+  virtual ProtoResult computeResult(
+      [[maybe_unused]] bool requestLaziness) override;
 
   VariableToColumnMap computeVariableToColumnMap() const override;
 };

src/engine/ExportQueryExecutionTrees.cpp

@@ -13,26 +13,48 @@
 #include "util/http/MediaTypes.h"
 
 // __________________________________________________________________________
-namespace {
+
+cppcoro::generator<const IdTable&> ExportQueryExecutionTrees::getIdTables(
+    const Result& result) {
+  if (result.isDataEvaluated()) {
+    co_yield result.idTable();
+  } else {
+    for (const IdTable& idTable : result.idTables()) {
+      co_yield idTable;
+    }
+  }
+}
+
 // Return a range that contains the indices of the rows that have to be exported
 // from the `idTable` given the `LimitOffsetClause`. It takes into account the
 // LIMIT, the OFFSET, and the actual size of the `idTable`
-auto getRowIndices(const LimitOffsetClause& limitOffset, const Result& result) {
-  const IdTable& idTable = result.idTable();
-  return std::views::iota(limitOffset.actualOffset(idTable.size()),
-                          limitOffset.upperBound(idTable.size()));
+cppcoro::generator<ExportQueryExecutionTrees::IndexWithTable>
+ExportQueryExecutionTrees::getRowIndices(LimitOffsetClause limitOffset,
+                                         const Result& result) {
+  for (const IdTable& idTable : getIdTables(result)) {
+    uint64_t currentOffset = limitOffset.actualOffset(idTable.numRows());
+    uint64_t upperBound = limitOffset.upperBound(idTable.numRows());
+    for (size_t index = currentOffset; index < upperBound; index++) {
+      co_yield {index, idTable};
+    }
+    limitOffset._offset -= currentOffset;
+    if (limitOffset._limit.has_value()) {
+      limitOffset._limit =
+          limitOffset._limit.value() - (upperBound - currentOffset);
+    }
+  }
 }
-}  // namespace
 
 // _____________________________________________________________________________
 cppcoro::generator<QueryExecutionTree::StringTriple>
 ExportQueryExecutionTrees::constructQueryResultToTriples(
     const QueryExecutionTree& qet,
     const ad_utility::sparql_types::Triples& constructTriples,
-    LimitOffsetClause limitAndOffset, std::shared_ptr<const Result> res,
+    LimitOffsetClause limitAndOffset, std::shared_ptr<const Result> result,
     CancellationHandle cancellationHandle) {
-  for (size_t i : getRowIndices(limitAndOffset, *res)) {
-    ConstructQueryExportContext context{i, *res, qet.getVariableColumns(),
+  for (auto [i, idTable] : getRowIndices(limitAndOffset, *result)) {
+    ConstructQueryExportContext context{i, idTable, result->localVocab(),
+                                        qet.getVariableColumns(),
                                         qet.getQec()->getIndex()};
     using enum PositionInTriple;
     for (const auto& triple : constructTriples) {
@@ -111,10 +133,9 @@ nlohmann::json ExportQueryExecutionTrees::idTableToQLeverJSONArray(
     std::shared_ptr<const Result> result,
     CancellationHandle cancellationHandle) {
   AD_CORRECTNESS_CHECK(result != nullptr);
-  const IdTable& data = result->idTable();
   nlohmann::json json = nlohmann::json::array();
 
-  for (size_t rowIndex : getRowIndices(limitAndOffset, *result)) {
+  for (auto [rowIndex, idTable] : getRowIndices(limitAndOffset, *result)) {
     // We need the explicit `array` constructor for the special case of zero
     // variables.
     json.push_back(nlohmann::json::array());
@@ -124,7 +145,7 @@ nlohmann::json ExportQueryExecutionTrees::idTableToQLeverJSONArray(
         row.emplace_back(nullptr);
         continue;
       }
-      const auto& currentId = data(rowIndex, opt->columnIndex_);
+      const auto& currentId = idTable(rowIndex, opt->columnIndex_);
       const auto& optionalStringAndXsdType = idToStringAndType(
           qet.getQec()->getIndex(), currentId, result->localVocab());
       if (!optionalStringAndXsdType.has_value()) {
@@ -231,7 +252,7 @@ ExportQueryExecutionTrees::idToStringAndType(const Index& index, Id id,
     return std::pair{escapeFunction(word.toStringRepresentation()), nullptr};
   };
   switch (id.getDatatype()) {
-    case Datatype::WordVocabIndex: {
+    case WordVocabIndex: {
       std::optional<string> entity =
           index.idToOptionalString(id.getWordVocabIndex());
       AD_CONTRACT_CHECK(entity.has_value());
@@ -290,8 +311,6 @@ nlohmann::json ExportQueryExecutionTrees::selectQueryResultToSparqlJSON(
 
   std::erase(columns, std::nullopt);
 
-  const IdTable& idTable = result->idTable();
-
   json resultJson;
   std::vector<std::string> selectedVars =
       selectClause.getSelectedVariablesAsStrings();
@@ -359,7 +378,7 @@ nlohmann::json ExportQueryExecutionTrees::selectQueryResultToSparqlJSON(
     return b;
   };
 
-  for (size_t rowIndex : getRowIndices(limitAndOffset, *result)) {
+  for (auto [rowIndex, idTable] : getRowIndices(limitAndOffset, *result)) {
     // TODO: ordered_json` entries are ordered alphabetically, but insertion
     // order would be preferable.
     nlohmann::ordered_json binding;
@@ -432,10 +451,9 @@ ExportQueryExecutionTrees::selectQueryResultToStream(
   auto selectedColumnIndices =
       qet.selectedVariablesToColumnIndices(selectClause, true);
 
-  const auto& idTable = result->idTable();
   // special case : binary export of IdTable
   if constexpr (format == MediaType::octetStream) {
-    for (size_t i : getRowIndices(limitAndOffset, *result)) {
+    for (auto [i, idTable] : getRowIndices(limitAndOffset, *result)) {
       for (const auto& columnIndex : selectedColumnIndices) {
         if (columnIndex.has_value()) {
           co_yield std::string_view{reinterpret_cast<const char*>(&idTable(
@@ -463,7 +481,7 @@ ExportQueryExecutionTrees::selectQueryResultToStream(
   constexpr auto& escapeFunction = format == MediaType::tsv
                                        ? RdfEscaping::escapeForTsv
                                        : RdfEscaping::escapeForCsv;
-  for (size_t i : getRowIndices(limitAndOffset, *result)) {
+  for (auto [i, idTable] : getRowIndices(limitAndOffset, *result)) {
     for (size_t j = 0; j < selectedColumnIndices.size(); ++j) {
       if (selectedColumnIndices[j].has_value()) {
         const auto& val = selectedColumnIndices[j].value();
@@ -583,11 +601,10 @@ ad_utility::streams::stream_generator ExportQueryExecutionTrees::
   co_yield "\n<results>";
 
   result->logResultSize();
-  const auto& idTable = result->idTable();
   auto selectedColumnIndices =
       qet.selectedVariablesToColumnIndices(selectClause, false);
   // TODO<joka921> we could prefilter for the nonexisting variables.
-  for (size_t i : getRowIndices(limitAndOffset, *result)) {
+  for (auto [i, idTable] : getRowIndices(limitAndOffset, *result)) {
     co_yield "\n  <result>";
     for (size_t j = 0; j < selectedColumnIndices.size(); ++j) {
       if (selectedColumnIndices[j].has_value()) {
@@ -644,11 +661,16 @@ nlohmann::json ExportQueryExecutionTrees::computeQueryResultAsQLeverJSON(
     const ParsedQuery& query, const QueryExecutionTree& qet,
     const ad_utility::Timer& requestTimer,
     CancellationHandle cancellationHandle) {
-  std::shared_ptr<const Result> result = qet.getResult();
+  std::shared_ptr<const Result> result =
+      qet.getResult(query._limitOffset._limit.has_value());
   result->logResultSize();
+  // TODO<RobinTF> this timer only makes sense for non lazy results.
   auto timeResultComputation = requestTimer.msecs();
 
-  size_t resultSize = result->idTable().size();
+  std::optional<size_t> resultSize =
+      query.hasSelectClause() && result->isDataEvaluated()
+          ? std::optional{result->idTable().size()}
+          : std::nullopt;
 
   nlohmann::json j;
 
@@ -680,7 +702,7 @@ nlohmann::json ExportQueryExecutionTrees::computeQueryResultAsQLeverJSON(
                   qet, query.constructClause().triples_, query._limitOffset,
                   std::move(result), std::move(cancellationHandle));
   }
-  j["resultsize"] = query.hasSelectClause() ? resultSize : j["res"].size();
+  j["resultsize"] = resultSize.value_or(j["res"].size());
   j["time"]["total"] = std::to_string(requestTimer.msecs().count()) + "ms";
   j["time"]["computeResult"] =
       std::to_string(timeResultComputation.count()) + "ms";
@@ -694,15 +716,14 @@ ExportQueryExecutionTrees::computeResultAsStream(
     const ParsedQuery& parsedQuery, const QueryExecutionTree& qet,
     ad_utility::MediaType mediaType, CancellationHandle cancellationHandle) {
   auto compute = [&]<MediaType format> {
-    auto limitAndOffset = parsedQuery._limitOffset;
     return parsedQuery.hasSelectClause()
-               ? ExportQueryExecutionTrees::selectQueryResultToStream<format>(
-                     qet, parsedQuery.selectClause(), limitAndOffset,
+               ? selectQueryResultToStream<format>(
+                     qet, parsedQuery.selectClause(), parsedQuery._limitOffset,
                      std::move(cancellationHandle))
-               : ExportQueryExecutionTrees::constructQueryResultToStream<
-                     format>(qet, parsedQuery.constructClause().triples_,
-                             limitAndOffset, qet.getResult(),
-                             std::move(cancellationHandle));
+               : constructQueryResultToStream<format>(
+                     qet, parsedQuery.constructClause().triples_,
+                     parsedQuery._limitOffset, qet.getResult(),
+                     std::move(cancellationHandle));
   };
 
   using enum MediaType;

src/engine/ExportQueryExecutionTrees.h

@@ -177,4 +177,28 @@ class ExportQueryExecutionTrees {
       const QueryExecutionTree& qet,
       const parsedQuery::SelectClause& selectClause,
       LimitOffsetClause limitAndOffset, CancellationHandle cancellationHandle);
+
+  struct IndexWithTable {
+    size_t index_;
+    const IdTable& idTable_;
+  };
+
+  static cppcoro::generator<const IdTable&> getIdTables(const Result& result);
+  // Return a range that contains the indices of the rows that have to be
+  // exported from the `idTable` given the `LimitOffsetClause`. It takes into
+  // account the LIMIT, the OFFSET, and the actual size of the `idTable`
+  static cppcoro::generator<IndexWithTable> getRowIndices(
+      LimitOffsetClause limitOffset, const Result& result);
+
+  FRIEND_TEST(ExportQueryExecutionTrees, getIdTablesReturnsSingletonIterator);
+  FRIEND_TEST(ExportQueryExecutionTrees, getIdTablesMirrorsGenerator);
+  FRIEND_TEST(ExportQueryExecutionTrees, ensureCorrectSlicingOfSingleIdTable);
+  FRIEND_TEST(ExportQueryExecutionTrees,
+              ensureCorrectSlicingOfIdTablesWhenFirstIsSkipped);
+  FRIEND_TEST(ExportQueryExecutionTrees,
+              ensureCorrectSlicingOfIdTablesWhenLastIsSkipped);
+  FRIEND_TEST(ExportQueryExecutionTrees,
+              ensureCorrectSlicingOfIdTablesWhenFirstAndSecondArePartial);
+  FRIEND_TEST(ExportQueryExecutionTrees,
+              ensureCorrectSlicingOfIdTablesWhenFirstAndLastArePartial);
 };

src/engine/Filter.cpp

@@ -43,41 +43,67 @@ string Filter::getDescriptor() const {
 }
 
 // _____________________________________________________________________________
-Result Filter::computeResult([[maybe_unused]] bool requestLaziness) {
+ProtoResult Filter::computeResult(bool requestLaziness) {
   LOG(DEBUG) << "Getting sub-result for Filter result computation..." << endl;
-  std::shared_ptr<const Result> subRes = _subtree->getResult();
+  std::shared_ptr<const Result> subRes = _subtree->getResult(requestLaziness);
   LOG(DEBUG) << "Filter result computation..." << endl;
   checkCancellation();
 
-  IdTable idTable{getExecutionContext()->getAllocator()};
-  idTable.setNumColumns(subRes->idTable().numColumns());
-
-  size_t width = idTable.numColumns();
-  CALL_FIXED_SIZE(width, &Filter::computeFilterImpl, this, &idTable, *subRes);
-  LOG(DEBUG) << "Filter result computation done." << endl;
-  checkCancellation();
+  if (subRes->isDataEvaluated()) {
+    sparqlExpression::EvaluationContext evaluationContext(
+        *getExecutionContext(), _subtree->getVariableColumns(),
+        subRes->idTable(), getExecutionContext()->getAllocator(),
+        subRes->localVocab(), cancellationHandle_);
+
+    // TODO<joka921> This should be a mandatory argument to the
+    // EvaluationContext constructor.
+    evaluationContext._columnsByWhichResultIsSorted = subRes->sortedBy();
+
+    size_t width = evaluationContext._inputTable.numColumns();
+    IdTable result = CALL_FIXED_SIZE(width, &Filter::computeFilterImpl, this,
+                                     evaluationContext);
+    LOG(DEBUG) << "Filter result computation done." << endl;
+    checkCancellation();
+
+    return {std::move(result), resultSortedOn(), subRes->getSharedLocalVocab()};
+  }
+  return {filterInChunks(subRes), resultSortedOn(),
+          subRes->getSharedLocalVocab()};
+}
 
-  return {std::move(idTable), resultSortedOn(), subRes->getSharedLocalVocab()};
+// _____________________________________________________________________________
+cppcoro::generator<IdTable> Filter::filterInChunks(
+    std::shared_ptr<const Result> subRes) {
+  for (const IdTable& idTable : subRes->idTables()) {
+    sparqlExpression::EvaluationContext evaluationContext(
+        *getExecutionContext(), _subtree->getVariableColumns(), idTable,
+        getExecutionContext()->getAllocator(), subRes->localVocab(),
+        cancellationHandle_);
+
+    // TODO<joka921> This should be a mandatory argument to the
+    // EvaluationContext constructor.
+    evaluationContext._columnsByWhichResultIsSorted = subRes->sortedBy();
+
+    size_t width = evaluationContext._inputTable.numColumns();
+    co_yield CALL_FIXED_SIZE(width, &Filter::computeFilterImpl, this,
+                             evaluationContext);
+    LOG(DEBUG) << "Filter result chunk done." << endl;
+    checkCancellation();
+  }
 }
 
 // _____________________________________________________________________________
 template <size_t WIDTH>
-void Filter::computeFilterImpl(IdTable* outputIdTable,
-                               const Result& inputResultTable) {
-  sparqlExpression::EvaluationContext evaluationContext(
-      *getExecutionContext(), _subtree->getVariableColumns(),
-      inputResultTable.idTable(), getExecutionContext()->getAllocator(),
-      inputResultTable.localVocab(), cancellationHandle_);
-
-  // TODO<joka921> This should be a mandatory argument to the EvaluationContext
-  // constructor.
-  evaluationContext._columnsByWhichResultIsSorted = inputResultTable.sortedBy();
+IdTable Filter::computeFilterImpl(
+    sparqlExpression::EvaluationContext& evaluationContext) {
+  IdTable idTable{getExecutionContext()->getAllocator()};
+  idTable.setNumColumns(evaluationContext._inputTable.numColumns());
 
   sparqlExpression::ExpressionResult expressionResult =
       _expression.getPimpl()->evaluate(&evaluationContext);
 
-  const auto input = inputResultTable.idTable().asStaticView<WIDTH>();
-  auto output = std::move(*outputIdTable).toStatic<WIDTH>();
+  const auto input = evaluationContext._inputTable.asStaticView<WIDTH>();
+  auto output = std::move(idTable).toStatic<WIDTH>();
   // Clang 17 seems to incorrectly deduce the type, so try to trick it
   std::remove_const_t<decltype(output)>& output2 = output;
 
@@ -123,7 +149,7 @@ void Filter::computeFilterImpl(IdTable* outputIdTable,
 
   std::visit(visitor, std::move(expressionResult));
 
-  *outputIdTable = std::move(output).toDynamic();
+  return std::move(output).toDynamic();
 }
 
 // _____________________________________________________________________________