Implement lazy join

src/engine/AddCombinedRowToTable.h

@@ -26,6 +26,7 @@ class AddCombinedRowToIdTable {
   size_t numJoinColumns_;
   std::optional<std::array<IdTableView<0>, 2>> inputLeftAndRight_;
   IdTable resultTable_;
+  LocalVocab mergedVocab_{};
 
   // This struct stores the information, which row indices from the input are
   // combined into a given row index in the output, i.e. "To obtain the
@@ -62,7 +63,7 @@ class AddCombinedRowToIdTable {
   // This callback is called with the result as an argument each time `flush()`
   // is called. It can be used to consume parts of the result early, before the
   // complete operation has finished.
-  using BlockwiseCallback = std::function<void(IdTable&)>;
+  using BlockwiseCallback = std::function<void(IdTable&, LocalVocab&)>;
   [[no_unique_address]] BlockwiseCallback blockwiseCallback_{ad_utility::noop};
 
   using CancellationHandle = ad_utility::SharedCancellationHandle;
@@ -138,10 +139,17 @@ class AddCombinedRowToIdTable {
         return table;
       }
     };
+    auto mergeVocab = [this]<typename T>(const T& table) {
+      if constexpr (requires { table.getLocalVocab(); }) {
+        mergedVocab_.mergeWith(std::span{&table.getLocalVocab(), 1});
+      }
+    };
     if (nextIndex_ != 0) {
       AD_CORRECTNESS_CHECK(inputLeftAndRight_.has_value());
       flush();
     }
+    mergeVocab(inputLeft);
+    mergeVocab(inputRight);
     inputLeftAndRight_ = std::array{toView(inputLeft), toView(inputRight)};
     checkNumColumns();
   }
@@ -188,6 +196,8 @@ class AddCombinedRowToIdTable {
     return std::move(resultTable_);
   }
 
+  LocalVocab& localVocab() { return mergedVocab_; }
+
   // Disable copying and moving, it is currently not needed and makes it harder
   // to reason about
   AddCombinedRowToIdTable(const AddCombinedRowToIdTable&) = delete;
@@ -320,7 +330,7 @@ class AddCombinedRowToIdTable {
     indexBuffer_.clear();
     optionalIndexBuffer_.clear();
     nextIndex_ = 0;
-    std::invoke(blockwiseCallback_, result);
+    std::invoke(blockwiseCallback_, result, mergedVocab_);
   }
   const IdTableView<0>& inputLeft() const {
     return inputLeftAndRight_.value()[0];

src/engine/Join.h

@@ -11,7 +11,7 @@
 #include "engine/QueryExecutionTree.h"
 #include "util/HashMap.h"
 #include "util/HashSet.h"
-#include "util/JoinAlgorithms/JoinAlgorithms.h"
+#include "util/TypeTraits.h"
 
 class Join : public Operation {
  private:
@@ -33,6 +33,7 @@ class Join : public Operation {
        std::shared_ptr<QueryExecutionTree> t2, ColumnIndex t1JoinCol,
        ColumnIndex t2JoinCol);
 
+  static constexpr size_t CHUNK_SIZE = 100'000;
   // A very explicit constructor, which initializes an invalid join object (it
   // has no subtrees, which violates class invariants). These invalid Join
   // objects can be used for unit tests that only test member functions which
@@ -93,6 +94,38 @@ class Join : public Operation {
   void join(const IdTable& a, ColumnIndex jc1, const IdTable& b,
             ColumnIndex jc2, IdTable* result) const;
 
+  // Helper function to compute the result of a join operation and conditionally
+  // return a lazy or fully materialized result depending on `requestLaziness`.
+  // This is achieved by running the `action` lambda in a separate thread and
+  // returning a lazy result that reads from the queue of the thread. If
+  // `requestLaziness` is false, the result is fully materialized and returned
+  // directly.
+  // `permutation` indicates a permutation to apply to the result columns before
+  // yielding/returning them. An empty vector means no permutation is applied.
+  // `action` is a lambda that can be used to send partial chunks to a consumer
+  // in addition to returning the remaining result. If laziness is not required
+  // it is a no-op.
+  ProtoResult createResult(
+      bool requestedLaziness,
+      ad_utility::InvocableWithExactReturnType<
+          Result::IdTableVocabPair,
+          std::function<void(IdTable&, LocalVocab&)>> auto action,
+      std::vector<ColumnIndex> permutation = {}) const;
+
+  // Helper function that cheaply checks if a join could contain undefined. For
+  // fully materialized tables it can just look at the first element. For lazy
+  // tables it has to look at the meta information which could potentially
+  // indicate undefinedness even when all values are defined.
+  static bool couldContainUndef(const auto& blocks, const auto& tree,
+                                ColumnIndex joinColumn);
+
+  // Fallback implementation of a join that is used when at least one of the two
+  // inputs is not fully materialized. This represents the general case where we
+  // don't have any optimization left to try.
+  ProtoResult lazyJoin(std::shared_ptr<const Result> a, ColumnIndex jc1,
+                       std::shared_ptr<const Result> b, ColumnIndex jc2,
+                       bool requestLaziness) const;
+
   /**
    * @brief Joins IdTables dynA and dynB on join column jc2, returning
    * the result in dynRes. Creates a cross product for matching rows by putting
@@ -113,24 +146,24 @@ class Join : public Operation {
   virtual string getCacheKeyImpl() const override;
 
  private:
-  ProtoResult computeResult([[maybe_unused]] bool requestLaziness) override;
+  ProtoResult computeResult(bool requestLaziness) override;
 
   VariableToColumnMap computeVariableToColumnMap() const override;
 
   // A special implementation that is called when both children are
   // `IndexScan`s. Uses the lazy scans to only retrieve the subset of the
   // `IndexScan`s that is actually needed without fully materializing them.
-  IdTable computeResultForTwoIndexScans();
+  ProtoResult computeResultForTwoIndexScans(bool requestLaziness) const;
 
   // A special implementation that is called when one of the children is an
   // `IndexScan`. The argument `scanIsLeft` determines whether the `IndexScan`
   // is the left or the right child of this `Join`. This needs to be known to
   // determine the correct order of the columns in the result.
   template <bool scanIsLeft>
-  IdTable computeResultForIndexScanAndIdTable(const IdTable& idTable,
-                                              ColumnIndex joinColTable,
-                                              IndexScan& scan,
-                                              ColumnIndex joinColScan);
+  ProtoResult computeResultForIndexScanAndIdTable(
+      bool requestLaziness, std::shared_ptr<const Result> resultWithIdTable,
+      ColumnIndex joinColTable, std::shared_ptr<IndexScan> scan,
+      ColumnIndex joinColScan) const;
 
   /*
    * @brief Combines 2 rows like in a join and inserts the result in the

src/engine/LocalVocab.h

@@ -97,6 +97,9 @@ class LocalVocab {
   // Return true if and only if the local vocabulary is empty.
   bool empty() const { return size() == 0; }
 
+  // The number of set stores (primary set and other sets).
+  size_t numSets() const { return 1 + otherWordSets_.size(); }
+
   // Get the `LocalVocabEntry` corresponding to the given `LocalVocabIndex`.
   //
   // NOTE: This used to be a more complex function but is now a simple

src/engine/Operation.cpp

@@ -120,6 +120,10 @@ ProtoResult Operation::runComputation(const ad_utility::Timer& timer,
   // correctly because the result was computed, so we can pass `nullopt` as
   // the last argument.
   if (result.isFullyMaterialized()) {
+    size_t numLocalVocabs = result.localVocab().numSets();
+    if (numLocalVocabs > 1) {
+      runtimeInfo().addDetail("num-local-vocabs", numLocalVocabs);
+    }
     updateRuntimeInformationOnSuccess(result.idTable().size(),
                                       ad_utility::CacheStatus::computed,
                                       timer.msecs(), std::nullopt);

src/index/IndexImpl.cpp

@@ -192,24 +192,24 @@ IndexImpl::buildOspWithPatterns(
         // them to the queue.
         IdTable outputBufferTable{NumColumnsIndexBuilding + 2,
                                   ad_utility::makeUnlimitedAllocator<Id>()};
-        auto pushToQueue =
-            [&, bufferSize =
-                    BUFFER_SIZE_JOIN_PATTERNS_WITH_OSP.load()](IdTable& table) {
-              if (table.numRows() >= bufferSize) {
-                if (!outputBufferTable.empty()) {
-                  queue.push(std::move(outputBufferTable));
-                  outputBufferTable.clear();
-                }
-                queue.push(std::move(table));
-              } else {
-                outputBufferTable.insertAtEnd(table.begin(), table.end());
-                if (outputBufferTable.size() >= bufferSize) {
-                  queue.push(std::move(outputBufferTable));
-                  outputBufferTable.clear();
-                }
-              }
-              table.clear();
-            };
+        auto pushToQueue = [&, bufferSize =
+                                   BUFFER_SIZE_JOIN_PATTERNS_WITH_OSP.load()](
+                               IdTable& table, LocalVocab&) {
+          if (table.numRows() >= bufferSize) {
+            if (!outputBufferTable.empty()) {
+              queue.push(std::move(outputBufferTable));
+              outputBufferTable.clear();
+            }
+            queue.push(std::move(table));
+          } else {
+            outputBufferTable.insertAtEnd(table.begin(), table.end());
+            if (outputBufferTable.size() >= bufferSize) {
+              queue.push(std::move(outputBufferTable));
+              outputBufferTable.clear();
+            }
+          }
+          table.clear();
+        };
 
         lazyOptionalJoinOnFirstColumn(ospAsBlocksTransformed, lazyPatternScan,
                                       pushToQueue);

src/util/JoinAlgorithms/JoinColumnMapping.h

@@ -8,6 +8,7 @@
 #include <cstdint>
 #include <vector>
 
+#include "engine/LocalVocab.h"
 #include "util/Algorithm.h"
 
 namespace ad_utility {
@@ -108,6 +109,7 @@ template <typename Table>
 struct IdTableAndFirstCol {
  private:
   Table table_;
+  LocalVocab localVocab_;
 
  public:
   // Typedef needed for generic interfaces.
@@ -116,7 +118,8 @@ struct IdTableAndFirstCol {
       std::decay_t<decltype(std::as_const(table_).getColumn(0).begin())>;
 
   // Construct by taking ownership of the table.
-  explicit IdTableAndFirstCol(Table t) : table_{std::move(t)} {}
+  IdTableAndFirstCol(Table t, LocalVocab localVocab)
+      : table_{std::move(t)}, localVocab_{std::move(localVocab)} {}
 
   // Get access to the first column.
   decltype(auto) col() { return table_.getColumn(0); }
@@ -131,6 +134,8 @@ struct IdTableAndFirstCol {
   bool empty() const { return col().empty(); }
 
   const Id& operator[](size_t idx) const { return col()[idx]; }
+  const Id& front() const { return col().front(); }
+  const Id& back() const { return col().back(); }
 
   size_t size() const { return col().size(); }
 
@@ -141,5 +146,7 @@ struct IdTableAndFirstCol {
   IdTableView<I> asStaticView() const {
     return table_.template asStaticView<I>();
   }
+
+  const LocalVocab& getLocalVocab() const { return localVocab_; }
 };
 }  // namespace ad_utility