Move all batch evaluation logic to ConstructBatchProcessor

Author: marvin7122

src/engine/ConstructBatchProcessor.cpp

@@ -100,6 +100,70 @@ std::optional<std::string> ConstructBatchProcessor::processCurrentRow() {
   return std::nullopt;
 }
 
+// _____________________________________________________________________________
+std::optional<ConstructBatchProcessor::StringTriple>
+ConstructBatchProcessor::getStringTriple() {
+  while (batchStart_ < rowIndicesVec_.size()) {
+    blueprint_->cancellationHandle_->throwIfCancelled();
+
+    loadBatchIfNeeded();
+
+    if (auto result = processCurrentBatchAsStringTriple()) {
+      return result;
+    }
+
+    advanceToNextBatch();
+  }
+
+  return std::nullopt;
+}
+
+// _____________________________________________________________________________
+std::optional<ConstructBatchProcessor::StringTriple>
+ConstructBatchProcessor::processCurrentBatchAsStringTriple() {
+  while (rowInBatchIdx_ < batchCache_->numRows_) {
+    if (auto result = processCurrentRowAsStringTriple()) {
+      return result;
+    }
+    advanceToNextRow();
+  }
+  return std::nullopt;
+}
+
+// _____________________________________________________________________________
+std::optional<ConstructBatchProcessor::StringTriple>
+ConstructBatchProcessor::processCurrentRowAsStringTriple() {
+  while (tripleIdx_ < blueprint_->numTemplateTriples()) {
+    auto subject =
+        getTermStringPtr(tripleIdx_, 0, *batchCache_, rowInBatchIdx_);
+    auto predicate =
+        getTermStringPtr(tripleIdx_, 1, *batchCache_, rowInBatchIdx_);
+    auto object = getTermStringPtr(tripleIdx_, 2, *batchCache_, rowInBatchIdx_);
+
+    ++tripleIdx_;
+
+    StringTriple triple = instantiateTriple(subject, predicate, object);
+    if (!triple.isEmpty()) {
+      return triple;
+    }
+    // Triple was UNDEF (incomplete), continue to next triple pattern.
+  }
+
+  return std::nullopt;
+}
+
+// _____________________________________________________________________________
+ConstructBatchProcessor::StringTriple
+ConstructBatchProcessor::instantiateTriple(
+    const std::shared_ptr<const std::string>& subject,
+    const std::shared_ptr<const std::string>& predicate,
+    const std::shared_ptr<const std::string>& object) const {
+  if (!subject || !predicate || !object) {
+    return StringTriple{};
+  }
+  return StringTriple{*subject, *predicate, *object};
+}
+
 // _____________________________________________________________________________
 void ConstructBatchProcessor::advanceToNextRow() {
   ++rowInBatchIdx_;

src/engine/ConstructBatchProcessor.h

@@ -15,6 +15,7 @@
 #include "backports/span.h"
 #include "engine/ConstructIdCache.h"
 #include "engine/InstantiationBlueprint.h"
+#include "engine/QueryExecutionTree.h"
 #include "engine/QueryExportTypes.h"
 #include "util/http/MediaTypes.h"
 #include "util/stream_generator.h"
@@ -27,6 +28,7 @@ class ConstructBatchProcessor
  public:
   using IdCache = ConstructIdCache;
   using IdCacheStatsLogger = ConstructIdCacheStatsLogger;
+  using StringTriple = QueryExecutionTree::StringTriple;
 
   // Default batch size for processing rows.
   static constexpr size_t DEFAULT_BATCH_SIZE = 64;
@@ -46,6 +48,10 @@ class ConstructBatchProcessor
   // Returns the next formatted triple, or nullopt when exhausted.
   std::optional<std::string> get() override;
 
+  // Returns the next triple as StringTriple, or nullopt when exhausted.
+  // Use either get() OR getStringTriple(), not both on the same instance.
+  std::optional<StringTriple> getStringTriple();
+
  private:
   // Evaluates all `Variables` and `BlankNodes` for a batch of result-table
   // rows. Column-oriented access pattern for variables.
@@ -88,6 +94,13 @@ class ConstructBatchProcessor
       const std::shared_ptr<const std::string>& predicate,
       const std::shared_ptr<const std::string>& object) const;
 
+  // Instantiates a single triple as StringTriple. Returns empty StringTriple
+  // if any component is UNDEF.
+  StringTriple instantiateTriple(
+      const std::shared_ptr<const std::string>& subject,
+      const std::shared_ptr<const std::string>& predicate,
+      const std::shared_ptr<const std::string>& object) const;
+
   // Load a new batch of rows for evaluation if we don't have one.
   void loadBatchIfNeeded();
 
@@ -97,6 +110,12 @@ class ConstructBatchProcessor
   // Process triples for the current row, returning the next formatted triple.
   std::optional<std::string> processCurrentRow();
 
+  // Process rows in the current batch, returning the next `StringTriple`.
+  std::optional<StringTriple> processCurrentBatchAsStringTriple();
+
+  // Process triples for the current row, returning the next `StringTriple`.
+  std::optional<StringTriple> processCurrentRowAsStringTriple();
+
   // Advance to next row in batch.
   void advanceToNextRow();
 
@@ -126,4 +145,23 @@ class ConstructBatchProcessor
   std::optional<BatchEvaluationCache> batchCache_;
 };
 
+// _____________________________________________________________________________
+// Adapter to use ConstructBatchProcessor::getStringTriple() with
+// InputRangeTypeErased<StringTriple>. This is a thin wrapper that delegates
+// to the batch processor's getStringTriple() method.
+class StringTripleBatchProcessor : public ad_utility::InputRangeFromGet<
+                                       ConstructBatchProcessor::StringTriple> {
+ public:
+  explicit StringTripleBatchProcessor(
+      std::unique_ptr<ConstructBatchProcessor> processor)
+      : processor_(std::move(processor)) {}
+
+  std::optional<ConstructBatchProcessor::StringTriple> get() override {
+    return processor_->getStringTriple();
+  }
+
+ private:
+  std::unique_ptr<ConstructBatchProcessor> processor_;
+};
+
 #endif  // QLEVER_SRC_ENGINE_CONSTRUCTBATCHPROCESSOR_H