diff --git a/src/engine/Filter.h b/src/engine/Filter.h
index 2ac01e1845..1785c38670 100644
--- a/src/engine/Filter.h
+++ b/src/engine/Filter.h
@@ -54,6 +54,11 @@ class Filter : public Operation {
     return _subtree->getMultiplicity(col);
   }
 
+  std::vector<Operation*> getIndexScansForSortVariables(
+      std::span<const Variable> variables) override {
+    return _subtree->getIndexScansForSortVariables(variables);
+  }
+
  private:
   VariableToColumnMap computeVariableToColumnMap() const override {
     return _subtree->getVariableColumns();
diff --git a/src/engine/IndexScan.cpp b/src/engine/IndexScan.cpp
index 1e6b71981d..211409305d 100644
--- a/src/engine/IndexScan.cpp
+++ b/src/engine/IndexScan.cpp
@@ -232,6 +232,11 @@ IdTable IndexScan::materializedIndexScan() const {
 // _____________________________________________________________________________
 ProtoResult IndexScan::computeResult(bool requestLaziness) {
   LOG(DEBUG) << "IndexScan result computation...\n";
+  if (explicitLazyResult_.has_value()) {
+    AD_CORRECTNESS_CHECK(requestLaziness);
+    absl::Cleanup cleanup{[this]() { explicitLazyResult_.reset(); }};
+    return {std::move(explicitLazyResult_.value()), resultSortedOn()};
+  }
   if (requestLaziness) {
     return {chunkedIndexScan(), resultSortedOn()};
   }
@@ -339,7 +344,14 @@ IndexScan::getBlockMetadata() const {
 // _____________________________________________________________________________
 std::optional<std::vector<CompressedBlockMetadata>>
 IndexScan::getBlockMetadataOptionallyPrefiltered() const {
-  auto optBlockSpan = getBlockMetadata();
+  auto optBlockSpan =
+      [&]() -> std::optional<std::span<const CompressedBlockMetadata>> {
+    if (prefilteredBlocks_.has_value()) {
+      return prefilteredBlocks_.value();
+    } else {
+      return getBlockMetadata();
+    }
+  }();
   std::optional<std::vector<CompressedBlockMetadata>> optBlocks = std::nullopt;
   if (optBlockSpan.has_value()) {
     const auto& blockSpan = optBlockSpan.value();
@@ -437,6 +449,34 @@ IndexScan::lazyScanForJoinOfTwoScans(const IndexScan& s1, const IndexScan& s2) {
   return result;
 }
 
+// _____________________________________________________________________________
+void IndexScan::setBlocksForJoinOfIndexScans(Operation* left,
+                                             Operation* right) {
+  auto& leftScan = dynamic_cast<IndexScan&>(*left);
+  auto& rightScan = dynamic_cast<IndexScan&>(*right);
+
+  auto getBlocks = [](IndexScan& scan) {
+    auto metaBlocks = scan.getMetadataForScan();
+    if (!metaBlocks.has_value()) {
+      return metaBlocks;
+    }
+    if (scan.prefilteredBlocks_.has_value()) {
+      metaBlocks.value().blockMetadata_ = scan.prefilteredBlocks_.value();
+    }
+    return metaBlocks;
+  };
+
+  auto metaBlocks1 = getBlocks(leftScan);
+  auto metaBlocks2 = getBlocks(rightScan);
+  if (!metaBlocks1.has_value() || !metaBlocks2.has_value()) {
+    return;
+  }
+  auto [blocks1, blocks2] = CompressedRelationReader::getBlocksForJoin(
+      metaBlocks1.value(), metaBlocks2.value());
+  leftScan.prefilteredBlocks_ = std::move(blocks1);
+  rightScan.prefilteredBlocks_ = std::move(blocks2);
+}
+
 // _____________________________________________________________________________
 Permutation::IdTableGenerator IndexScan::lazyScanForJoinOfColumnWithScan(
     std::span<const Id> joinColumn) const {
@@ -652,3 +692,21 @@ std::pair<Result::Generator, Result::Generator> IndexScan::prefilterTables(
   return {createPrefilteredJoinSide(state),
           createPrefilteredIndexScanSide(state)};
 }
+
+// _____________________________________________________________________________
+std::vector<Operation*> IndexScan::getIndexScansForSortVariables(
+    std::span<const Variable> variables) {
+  const auto& sorted = resultSortedOn();
+  if (resultSortedOn().size() < variables.size()) {
+    return {};
+  }
+  const auto& varColMap = getExternallyVisibleVariableColumns();
+  for (size_t i = 0; i < variables.size(); ++i) {
+    auto it = varColMap.find(variables[i]);
+    if (it == varColMap.end() ||
+        it->second.columnIndex_ != resultSortedOn().at(i)) {
+      return {};
+    }
+  }
+  return {this};
+}
diff --git a/src/engine/IndexScan.h b/src/engine/IndexScan.h
index c10680f59e..a23a5a1a5f 100644
--- a/src/engine/IndexScan.h
+++ b/src/engine/IndexScan.h
@@ -36,6 +36,9 @@ class IndexScan final : public Operation {
   std::vector<ColumnIndex> additionalColumns_;
   std::vector<Variable> additionalVariables_;
 
+  std::optional<std::vector<CompressedBlockMetadata>> prefilteredBlocks_;
+  std::optional<Result::Generator> explicitLazyResult_ = std::nullopt;
+
  public:
   IndexScan(QueryExecutionContext* qec, Permutation::Enum permutation,
             const SparqlTriple& triple, Graphs graphsToFilter = std::nullopt,
@@ -108,6 +111,12 @@ class IndexScan final : public Operation {
   std::pair<Result::Generator, Result::Generator> prefilterTables(
       Result::Generator input, ColumnIndex joinColumn);
 
+  static void setBlocksForJoinOfIndexScans(Operation* left, Operation* right);
+
+  void setLazyResultManually(Result::Generator lazyResult) {
+    explicitLazyResult_ = std::move(lazyResult);
+  }
+
  private:
   // Implementation detail that allows to consume a generator from two other
   // cooperating generators. Needs to be forward declared as it is used by
@@ -234,4 +243,14 @@ class IndexScan final : public Operation {
   Permutation::IdTableGenerator getLazyScan(
       std::vector<CompressedBlockMetadata> blocks) const;
   std::optional<Permutation::MetadataAndBlocks> getMetadataForScan() const;
+
+  // TODO<joka921> Comment.
+  void setPrefilteredBlocks(
+      std::vector<CompressedBlockMetadata> prefilteredBlocks) {
+    prefilteredBlocks_ = std::move(prefilteredBlocks);
+    disableCaching();
+  }
+
+  std::vector<Operation*> getIndexScansForSortVariables(
+      std::span<const Variable> variables) override;
 };
diff --git a/src/engine/Join.cpp b/src/engine/Join.cpp
index d3c5370e16..f380e8a7e3 100644
--- a/src/engine/Join.cpp
+++ b/src/engine/Join.cpp
@@ -181,6 +181,14 @@ ProtoResult Join::computeResult(bool requestLaziness) {
   auto rightResIfCached = getCachedOrSmallResult(*_right);
   checkCancellation();
 
+  std::span joinVarSpan{&_joinVar, 1};
+  auto leftIndexScans = _left->getIndexScansForSortVariables(joinVarSpan);
+  auto rightIndexScans = _right->getIndexScansForSortVariables(joinVarSpan);
+  for (auto* left : leftIndexScans) {
+    for (auto* right : rightIndexScans) {
+      IndexScan::setBlocksForJoinOfIndexScans(left, right);
+    }
+  }
   auto leftIndexScan =
       std::dynamic_pointer_cast<IndexScan>(_left->getRootOperation());
   if (leftIndexScan &&
@@ -189,9 +197,6 @@ ProtoResult Join::computeResult(bool requestLaziness) {
       AD_CORRECTNESS_CHECK(rightResIfCached->isFullyMaterialized());
       return computeResultForIndexScanAndIdTable<true>(
           requestLaziness, std::move(rightResIfCached), leftIndexScan);
-
-    } else if (!leftResIfCached) {
-      return computeResultForTwoIndexScans(requestLaziness);
     }
   }
 
@@ -647,47 +652,6 @@ void Join::addCombinedRowToIdTable(const ROW_A& rowA, const ROW_B& rowB,
   }
 }
 
-// ______________________________________________________________________________________________________
-ProtoResult Join::computeResultForTwoIndexScans(bool requestLaziness) const {
-  return createResult(
-      requestLaziness,
-      [this](std::function<void(IdTable&, LocalVocab&)> yieldTable) {
-        auto leftScan =
-            std::dynamic_pointer_cast<IndexScan>(_left->getRootOperation());
-        auto rightScan =
-            std::dynamic_pointer_cast<IndexScan>(_right->getRootOperation());
-        AD_CORRECTNESS_CHECK(leftScan && rightScan);
-        // The join column already is the first column in both inputs, so we
-        // don't have to permute the inputs and results for the
-        // `AddCombinedRowToIdTable` class to work correctly.
-        AD_CORRECTNESS_CHECK(_leftJoinCol == 0 && _rightJoinCol == 0);
-        auto rowAdder = makeRowAdder(std::move(yieldTable));
-
-        ad_utility::Timer timer{
-            ad_utility::timer::Timer::InitialStatus::Started};
-        auto [leftBlocksInternal, rightBlocksInternal] =
-            IndexScan::lazyScanForJoinOfTwoScans(*leftScan, *rightScan);
-        runtimeInfo().addDetail("time-for-filtering-blocks", timer.msecs());
-
-        auto leftBlocks = convertGenerator(std::move(leftBlocksInternal));
-        auto rightBlocks = convertGenerator(std::move(rightBlocksInternal));
-
-        ad_utility::zipperJoinForBlocksWithoutUndef(leftBlocks, rightBlocks,
-                                                    std::less{}, rowAdder);
-
-        leftScan->updateRuntimeInfoForLazyScan(leftBlocks.details());
-        rightScan->updateRuntimeInfoForLazyScan(rightBlocks.details());
-
-        AD_CORRECTNESS_CHECK(leftBlocks.details().numBlocksRead_ <=
-                             rightBlocks.details().numElementsRead_);
-        AD_CORRECTNESS_CHECK(rightBlocks.details().numBlocksRead_ <=
-                             leftBlocks.details().numElementsRead_);
-        auto localVocab = std::move(rowAdder.localVocab());
-        return Result::IdTableVocabPair{std::move(rowAdder).resultTable(),
-                                        std::move(localVocab)};
-      });
-}
-
 // ______________________________________________________________________________________________________
 template <bool idTableIsRightInput>
 ProtoResult Join::computeResultForIndexScanAndIdTable(
diff --git a/src/engine/Join.h b/src/engine/Join.h
index 8c8978c8d3..de159dd61e 100644
--- a/src/engine/Join.h
+++ b/src/engine/Join.h
@@ -149,11 +149,6 @@ class Join : public Operation {
 
   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.
-  ProtoResult computeResultForTwoIndexScans(bool requestLaziness) const;
-
   // A special implementation that is called when exactly one of the children is
   // an `IndexScan` and the other one is a fully materialized result. The
   // argument `idTableIsRightInput` determines whether the `IndexScan` is the
@@ -214,4 +209,13 @@ class Join : public Operation {
   // Helper function to create the commonly used instance of this class.
   ad_utility::AddCombinedRowToIdTable makeRowAdder(
       std::function<void(IdTable&, LocalVocab&)> callback) const;
+
+ public:
+  std::vector<Operation*> getIndexScansForSortVariables(
+      std::span<const Variable> variables) override {
+    auto result = _left->getIndexScansForSortVariables(variables);
+    auto right = _right->getIndexScansForSortVariables(variables);
+    result.insert(result.end(), right.begin(), right.end());
+    return result;
+  }
 };
diff --git a/src/engine/Operation.cpp b/src/engine/Operation.cpp
index cacba381ae..ab7f1ec9ed 100644
--- a/src/engine/Operation.cpp
+++ b/src/engine/Operation.cpp
@@ -190,8 +190,9 @@ CacheValue Operation::runComputationAndPrepareForCache(
     const ad_utility::Timer& timer, ComputationMode computationMode,
     const QueryCacheKey& cacheKey, bool pinned) {
   auto& cache = _executionContext->getQueryTreeCache();
+  bool suitableForCaching = isSuitableForCaching();
   auto result = runComputation(timer, computationMode);
-  if (!result.isFullyMaterialized() &&
+  if (suitableForCaching && !result.isFullyMaterialized() &&
       !unlikelyToFitInCache(cache.getMaxSizeSingleEntry())) {
     AD_CONTRACT_CHECK(!pinned);
     result.cacheDuringConsumption(
@@ -276,12 +277,16 @@ std::shared_ptr<const Result> Operation::getResult(
 
     bool onlyReadFromCache = computationMode == ComputationMode::ONLY_IF_CACHED;
 
-    auto result =
-        pinResult ? cache.computeOncePinned(cacheKey, cacheSetup,
-                                            onlyReadFromCache, suitedForCache)
-                  : cache.computeOnce(cacheKey, cacheSetup, onlyReadFromCache,
-                                      suitedForCache);
-
+    auto result = [&]() -> QueryResultCache::ResultAndCacheStatus {
+      auto compute = [&](auto&&... args) {
+        if (!isSuitableForCaching()) {
+          return cache.computeWithoutCache(AD_FWD(args)...);
+        }
+        return pinResult ? cache.computeOncePinned(AD_FWD(args)...)
+                         : cache.computeOnce(AD_FWD(args)...);
+      };
+      return compute(cacheKey, cacheSetup, onlyReadFromCache, suitedForCache);
+    }();
     if (result._resultPointer == nullptr) {
       AD_CORRECTNESS_CHECK(onlyReadFromCache);
       return nullptr;
diff --git a/src/engine/Operation.h b/src/engine/Operation.h
index 3e06a9498e..57c1f3b905 100644
--- a/src/engine/Operation.h
+++ b/src/engine/Operation.h
@@ -23,6 +23,7 @@
 
 // forward declaration needed to break dependencies
 class QueryExecutionTree;
+class IndexScan;
 
 enum class ComputationMode {
   FULLY_MATERIALIZED,
@@ -90,6 +91,8 @@ class Operation {
   // limit/offset is applied post computation.
   bool externalLimitApplied_ = false;
 
+  bool isSuitableForCaching_ = true;
+
  public:
   // Holds a `PrefilterExpression` with its corresponding `Variable`.
   using PrefilterVariablePair = sparqlExpression::PrefilterExprVariablePair;
@@ -162,7 +165,7 @@ class Operation {
   // Get a unique, not ambiguous string representation for a subtree.
   // This should act like an ID for each subtree.
   // Calls  `getCacheKeyImpl` and adds the information about the `LIMIT` clause.
-  virtual string getCacheKey() const final {
+  virtual std::string getCacheKey() const final {
     auto result = getCacheKeyImpl();
     if (_limit._limit.has_value()) {
       absl::StrAppend(&result, " LIMIT ", _limit._limit.value());
@@ -173,6 +176,11 @@ class Operation {
     return result;
   }
 
+  // If this function returns `false`, then the result of this `Operation` will
+  // never be stored in the cache.
+  virtual bool isSuitableForCaching() const { return isSuitableForCaching_; }
+  virtual void disableCaching() final { isSuitableForCaching_ = true; }
+
  private:
   // The individual implementation of `getCacheKey` (see above) that has to be
   // customized by every child class.
@@ -426,6 +434,11 @@ class Operation {
       RuntimeInformation::Status status =
           RuntimeInformation::Status::optimizedOut);
 
+  virtual std::vector<Operation*> getIndexScansForSortVariables(
+      [[maybe_unused]] std::span<const Variable> variables) {
+    return {};
+  }
+
  private:
   // Create the runtime information in case the evaluation of this operation has
   // failed.
diff --git a/src/engine/QueryExecutionTree.cpp b/src/engine/QueryExecutionTree.cpp
index c9496fe958..6a5f5b8319 100644
--- a/src/engine/QueryExecutionTree.cpp
+++ b/src/engine/QueryExecutionTree.cpp
@@ -11,6 +11,8 @@
 #include <string>
 #include <vector>
 
+#include "engine/Filter.h"
+#include "engine/IndexScan.h"
 #include "engine/Sort.h"
 #include "parser/RdfEscaping.h"
 
@@ -164,8 +166,26 @@ std::shared_ptr<QueryExecutionTree> QueryExecutionTree::createSortedTree(
   }
 
   QueryExecutionContext* qec = qet->getRootOperation()->getExecutionContext();
+  /*
+  bool isResortedFilter = false;
+  if (auto filter = dynamic_cast<const Filter*>(qet->rootOperation_.get())) {
+    if (dynamic_cast<const
+  IndexScan*>(filter->getSubtree()->rootOperation_.get())) { isResortedFilter =
+  true;
+    }
+  }
+  */
   auto sort = std::make_shared<Sort>(qec, std::move(qet), sortColumns);
-  return std::make_shared<QueryExecutionTree>(qec, std::move(sort));
+  auto result = std::make_shared<QueryExecutionTree>(qec, std::move(sort));
+  /*
+  // TODO<joka921> This currently is a hack for the pubchem queries, let's
+  // discuss the options to do this.
+  if (isResortedFilter) {
+    result->getSizeEstimate();
+    result->sizeEstimate_.value() *= 20;
+  }
+  */
+  return result;
 }
 
 // _____________________________________________________________________________
diff --git a/src/engine/QueryExecutionTree.h b/src/engine/QueryExecutionTree.h
index 0eac785f16..ceea9b8efe 100644
--- a/src/engine/QueryExecutionTree.h
+++ b/src/engine/QueryExecutionTree.h
@@ -102,6 +102,18 @@ class QueryExecutionTree {
   setPrefilterGetUpdatedQueryExecutionTree(
       std::vector<Operation::PrefilterVariablePair> prefilterPairs) const;
 
+  virtual std::vector<Operation*> getIndexScansForSortVariables(
+      std::span<const Variable> variables) final {
+    auto result = rootOperation_->getIndexScansForSortVariables(variables);
+    if (result.empty()) {
+      return result;
+    }
+    rootOperation_->disableCaching();
+    cachedResult_.reset();
+    sizeEstimate_.reset();
+    return result;
+  }
+
   size_t getDistinctEstimate(size_t col) const {
     return static_cast<size_t>(rootOperation_->getSizeEstimate() /
                                rootOperation_->getMultiplicity(col));
diff --git a/src/engine/Sort.h b/src/engine/Sort.h
index d94a69c199..13aeaa7c4f 100644
--- a/src/engine/Sort.h
+++ b/src/engine/Sort.h
@@ -5,6 +5,8 @@
 
 #pragma once
 
+#include "engine/Filter.h"
+#include "engine/IndexScan.h"
 #include "engine/Operation.h"
 #include "engine/QueryExecutionTree.h"
 
@@ -53,7 +55,22 @@ class Sort : public Operation {
     // Return  at least 1, s.t. the query planner will never emit an unnecessary
     // sort of an empty `IndexScan`. This makes the testing of the query
     // planner much easier.
-    return std::max(1UL, nlogn + subcost);
+    size_t sizeEstimateOfFilteredScan = 0;
+    if (auto filter =
+            dynamic_cast<const Filter*>(subtree_->getRootOperation().get())) {
+      if (dynamic_cast<const IndexScan*>(
+              filter->getSubtree()->getRootOperation().get())) {
+        sizeEstimateOfFilteredScan = subtree_->getSizeEstimate();
+      }
+    }
+    auto result = std::max(1UL, nlogn + subcost);
+    // TODO<joka921> proper constants.
+    if (sizeEstimateOfFilteredScan > 1'000'000) {
+      result *= 10'000;
+    } else if (sizeEstimateOfFilteredScan > 0) {
+      result *= 20;
+    }
+    return result;
   }
 
   virtual bool knownEmptyResult() override {
diff --git a/src/index/CompressedRelation.h b/src/index/CompressedRelation.h
index 36cbf14af0..7f20527ad9 100644
--- a/src/index/CompressedRelation.h
+++ b/src/index/CompressedRelation.h
@@ -466,7 +466,7 @@ class CompressedRelationReader {
   // to be performed.
   struct ScanSpecAndBlocks {
     ScanSpecification scanSpec_;
-    const std::span<const CompressedBlockMetadata> blockMetadata_;
+    std::span<const CompressedBlockMetadata> blockMetadata_;
   };
 
   // This struct additionally contains the first and last triple of the scan
diff --git a/src/util/ConcurrentCache.h b/src/util/ConcurrentCache.h
index 2f22efde8c..40d87ee444 100644
--- a/src/util/ConcurrentCache.h
+++ b/src/util/ConcurrentCache.h
@@ -208,6 +208,22 @@ class ConcurrentCache {
                            suitedForCache);
   }
 
+  // Simply compute the result, but neither read it from the cache nor store it
+  // in cache. This function is there to have a unified interface with the above
+  // two functions.
+  ResultAndCacheStatus computeWithoutCache(
+      [[maybe_unused]] const Key& key,
+      const InvocableWithConvertibleReturnType<Value> auto& computeFunction,
+      bool onlyReadFromCache,
+      [[maybe_unused]] const InvocableWithConvertibleReturnType<
+          bool, const Value&> auto& suitedForCache) {
+    if (onlyReadFromCache) {
+      return {nullptr, CacheStatus::notInCacheAndNotComputed};
+    }
+    auto value = std::make_shared<Value>(computeFunction());
+    return {std::move(value), CacheStatus::computed};
+  }
+
   // Insert `value` into the cache, if the `key` is not already present. In case
   // `pinned` is true and the key is already present, the existing value is
   // pinned in case it is not pinned yet.