Prepare recomputation of triple statistics when there have been updates (ad-freiburg#2640)

Implement the functions for recomputing the triple statistics, as preparation for ad-freiburg#2408 (rebuild index)

NOTE: This could also be used to recompute the triple statistics stored in the `.meta-data.json` file, either periodically in the background, or when they are requested via `cmd=stats`. That is work for a separate PR

Author: RobinTF

src/index/IndexImpl.cpp

@@ -27,6 +27,7 @@
 #include "util/InputRangeUtils.h"
 #include "util/Iterators.h"
 #include "util/JoinAlgorithms/JoinAlgorithms.h"
+#include "util/ParallelExecutor.h"
 #include "util/ProgressBar.h"
 #include "util/ThreadSafeQueue.h"
 #include "util/Timer.h"
@@ -1787,3 +1788,107 @@ void IndexImpl::setPrefixesForEncodedValues(
   encodedIriManager_ =
       EncodedIriManager{std::move(prefixesWithoutAngleBrackets)};
 }
+
+// _____________________________________________________________________________
+void IndexImpl::countDistinct(std::optional<Id>& lastId, size_t& counter,
+                              const IdTable& table) {
+  AD_CORRECTNESS_CHECK(
+      !table.empty(), "Empty tables should never be yielded by the lazy scan.");
+  auto col = table.getColumn(0);
+  counter += ql::ranges::distance(col | ::ranges::views::unique([](Id a, Id b) {
+                                    return a.getBits() == b.getBits();
+                                  }));
+  if (lastId == col.front()) {
+    // Avoid double counting in case the last id of the previous block is the
+    // same as the first id of this block.
+    counter--;
+  }
+  lastId = col.back();
+}
+
+namespace {
+// Helper function that returns a packaged task that computes distinct counts
+// over all tables produced by scanning the given permutation. The customAction
+// is invoked for each table to allow for additional computations while
+// scanning.
+std::packaged_task<void()> computeStatistics(
+    const LocatedTriplesSharedState& locatedTriplesSharedState, size_t& counter,
+    const Permutation& permutation, auto customAction) {
+  return std::packaged_task{[&counter, &permutation, &locatedTriplesSharedState,
+                             customAction = std::move(customAction)]() {
+    auto cancellationHandle =
+        std::make_shared<ad_utility::SharedCancellationHandle::element_type>();
+    ScanSpecification scanSpec{std::nullopt, std::nullopt, std::nullopt};
+    auto tables = permutation.lazyScan(
+        permutation.getScanSpecAndBlocks(scanSpec, *locatedTriplesSharedState),
+        std::nullopt, CompressedRelationReader::ColumnIndicesRef{},
+        cancellationHandle, *locatedTriplesSharedState);
+    std::optional<Id> lastCol0 = std::nullopt;
+    for (const auto& table : tables) {
+      std::invoke(customAction, table);
+      IndexImpl::countDistinct(lastCol0, counter, table);
+    }
+  }};
+}
+}  // namespace
+
+// _____________________________________________________________________________
+nlohmann::json IndexImpl::recomputeStatistics(
+    const LocatedTriplesSharedState& locatedTriplesSharedState) const {
+  size_t numTriples = 0;
+  size_t numTriplesInternal = 0;
+  size_t numSubjects = 0;
+  size_t numPredicates = 0;
+  size_t numPredicatesInternal = 0;
+  size_t numObjects = 0;
+  uint64_t nextBlankNode = 0;
+
+  std::vector<std::packaged_task<void()>> tasks;
+
+  auto getCounterTask = [&locatedTriplesSharedState](
+                            size_t& counter, const Permutation& permutation,
+                            auto customAction) {
+    return computeStatistics(locatedTriplesSharedState, counter, permutation,
+                             customAction);
+  };
+
+  tasks.push_back(getCounterTask(
+      numPredicates, *pso_,
+      [&numTriples, &nextBlankNode](const IdTable& table) {
+        numTriples += table.numRows();
+        for (auto col : table.getColumns()) {
+          for (auto id : col) {
+            if (id.getDatatype() == Datatype::BlankNodeIndex) {
+              nextBlankNode =
+                  std::max(nextBlankNode, id.getBlankNodeIndex().get() + 1);
+            }
+          }
+        }
+      }));
+
+  tasks.push_back(getCounterTask(numPredicatesInternal,
+                                 pso_->internalPermutation(),
+                                 [&numTriplesInternal](const IdTable& table) {
+                                   numTriplesInternal += table.numRows();
+                                 }));
+
+  if (hasAllPermutations()) {
+    tasks.push_back(getCounterTask(numSubjects, *spo_, ad_utility::noop));
+    tasks.push_back(getCounterTask(numObjects, *osp_, ad_utility::noop));
+  }
+  ad_utility::runTasksInParallel(std::move(tasks));
+  auto configuration = configurationJson_;
+  configuration["num-triples"] =
+      NumNormalAndInternal{numTriples, numTriplesInternal};
+  configuration["num-predicates"] =
+      NumNormalAndInternal{numPredicates, numPredicatesInternal};
+  if (hasAllPermutations()) {
+    // These are unused.
+    AD_CORRECTNESS_CHECK(numSubjects_.internal == 0);
+    AD_CORRECTNESS_CHECK(numObjects_.internal == 0);
+    configuration["num-subjects"] = NumNormalAndInternal{numSubjects, 0};
+    configuration["num-objects"] = NumNormalAndInternal{numObjects, 0};
+  }
+  configuration["num-blank-nodes-total"] = nextBlankNode;
+  return configuration;
+}

src/index/IndexImpl.h

@@ -7,6 +7,8 @@
 #ifndef QLEVER_SRC_INDEX_INDEXIMPL_H
 #define QLEVER_SRC_INDEX_INDEXIMPL_H
 
+#include <gtest/gtest_prod.h>
+
 #include <memory>
 #include <optional>
 #include <string>
@@ -643,6 +645,7 @@ class IndexImpl {
   friend class IndexTest_createFromTsvTest_Test;
   friend class IndexTest_createFromOnDiskIndexTest_Test;
   friend class CreatePatternsFixture_createPatterns_Test;
+  FRIEND_TEST(IndexImpl, recomputeStatistics);
 
   bool isLiteral(std::string_view object) const;
 
@@ -801,6 +804,18 @@ class IndexImpl {
 
   void storeTextScoringParamsInConfiguration(TextScoringMetric scoringMetric,
                                              float b, float k);
+
+  // Helper function to count the number of distinct Ids in a sorted IdTable.
+  // `lastId` is used to keep track of the last seen Id between multiple calls
+  // for subsequent tables and `counter` is the counter that is incremented.
+  // This function is only exposed for testing.
+  static void countDistinct(std::optional<Id>& lastId, size_t& counter,
+                            const IdTable& table);
+
+  // Recompute the statistics about the index based on the passed located
+  // triples shared state.
+  nlohmann::json recomputeStatistics(
+      const LocatedTriplesSharedState& locatedTriplesSharedState) const;
 };
 
 #endif  // QLEVER_SRC_INDEX_INDEXIMPL_H

src/util/ParallelExecutor.h

@@ -0,0 +1,37 @@
+//   Copyright 2026 The QLever Authors, in particular:
+//
+//  2026 Robin Textor-Falconi <[email protected]>, UFR
+//
+//  UFR = University of Freiburg, Chair of Algorithms and Data Structures
+
+#ifndef QLEVER_SRC_UTIL_PARALLELEXECUTOR_H
+#define QLEVER_SRC_UTIL_PARALLELEXECUTOR_H
+
+#include <future>
+#include <vector>
+
+#include "util/jthread.h"
+
+namespace ad_utility {
+// Run the given tasks in parallel and wait for their completion. This function
+// will spawn a new thread for each task. If one of the tasks throws an
+// exception, this exception will be rethrown in the main thread. If multiple
+// tasks throw exceptions, only the first one will be rethrown.
+inline void runTasksInParallel(
+    std::vector<std::packaged_task<void()>>&& tasks) {
+  std::vector<std::future<void>> futures;
+  futures.reserve(tasks.size());
+  std::vector<JThread> threads;
+  futures.reserve(tasks.size());
+  for (auto& task : tasks) {
+    futures.push_back(task.get_future());
+    threads.push_back(JThread{std::move(task)});
+  }
+  // Wait for completion.
+  for (auto& future : futures) {
+    future.get();
+  }
+}
+}  // namespace ad_utility
+
+#endif  // QLEVER_SRC_UTIL_PARALLELEXECUTOR_H

test/CMakeLists.txt

@@ -488,3 +488,5 @@ addLinkAndDiscoverTest(UnitOfMeasurementTest util)
 addLinkAndDiscoverTest(MaterializedViewsTest qlever engine server)
 
 addLinkAndDiscoverTestNoLibs(ConstexprMapTest)
+
+addLinkAndDiscoverTestNoLibs(ParallelExecutorTest)

test/IndexTest.cpp

@@ -662,3 +662,81 @@ TEST(IndexTest, getBlankNodeManager) {
   const Index& index3 = getQec(kb)->getIndex();
   EXPECT_EQ(index3.getBlankNodeManager()->minIndex_, 3);
 }
+
+// _____________________________________________________________________________
+TEST(IndexImpl, recomputeStatistics) {
+  std::string turtleInput =
+      "<x> <label> \"alpha\" . <x> <label> \"A\" . "
+      "<y> <label> \"Beta\". <z> <label> \"zz\"@en";
+  auto index = makeTestIndex("recomputeStatistics", std::move(turtleInput));
+  auto cancellationHandle =
+      std::make_shared<ad_utility::SharedCancellationHandle::element_type>();
+
+  auto& indexImpl = index.getImpl();
+  // No-op, should return the same stats.
+  auto result = indexImpl.recomputeStatistics(
+      index.deltaTriplesManager().getCurrentLocatedTriplesSharedState());
+  EXPECT_EQ(result, indexImpl.configurationJson_);
+
+  // Now, modify the index by adding triples.
+  Id blankNodeId = Id::makeFromBlankNodeIndex(BlankNodeIndex::make(42));
+  index.deltaTriplesManager().modify<void>([&cancellationHandle, blankNodeId](
+                                               DeltaTriples& deltaTriples) {
+    LocalVocabEntry zzz{ad_utility::triple_component::Iri::fromIriref("<zzz>")};
+    LocalVocabEntry literal{
+        ad_utility::triple_component::Literal::fromStringRepresentation(
+            "\"test\"@en")};
+    Id zzzId = Id::makeFromLocalVocabIndex(&zzz);
+    Id literalId = Id::makeFromLocalVocabIndex(&literal);
+    // Create duplicate in different graph.
+    Id x = Id::makeFromVocabIndex(VocabIndex::make(11));
+    Id label = Id::makeFromVocabIndex(VocabIndex::make(10));
+    Id alpha = Id::makeFromVocabIndex(VocabIndex::make(1));
+    deltaTriples.insertTriples(
+        cancellationHandle, {IdTriple{{x, label, alpha, x}},
+                             IdTriple{{blankNodeId, zzzId, literalId, zzzId}}});
+  });
+
+  for (bool loadAllPermutations : {true, false}) {
+    using NNAI = Index::NumNormalAndInternal;
+
+    // Simulate scenario where not all permutations are loaded.
+    if (!loadAllPermutations) {
+      // Overwrite with unloaded permutation.
+      indexImpl.SPO() = Permutation{Permutation::SPO,
+                                    ad_utility::makeUnlimitedAllocator<Id>()};
+      // Zero out original values.
+      indexImpl.configurationJson_["num-subjects"] = NNAI(0, 0);
+      indexImpl.configurationJson_["num-objects"] = NNAI(0, 0);
+    }
+
+    auto newStats = indexImpl.recomputeStatistics(
+        index.deltaTriplesManager().getCurrentLocatedTriplesSharedState());
+    EXPECT_NE(newStats, indexImpl.configurationJson_);
+    EXPECT_EQ(newStats["num-triples"], NNAI(5, 6));
+    EXPECT_EQ(newStats["num-predicates"], NNAI(2, 4));
+    if (loadAllPermutations) {
+      EXPECT_EQ(newStats["num-subjects"], NNAI(4, 0));
+      EXPECT_EQ(newStats["num-objects"], NNAI(5, 0));
+    } else {
+      EXPECT_EQ(newStats["num-subjects"], NNAI(0, 0));
+      EXPECT_EQ(newStats["num-objects"], NNAI(0, 0));
+    }
+    // Blank node ids are remapped, so we cannot predict the exact number.
+    EXPECT_NE(newStats["num-blank-nodes-total"], 0);
+  }
+}
+
+// _____________________________________________________________________________
+TEST(IndexImpl, countDistinct) {
+  std::vector<IdTable> tables;
+  tables.push_back(makeIdTableFromVector({{1}, {2}}));
+  tables.push_back(makeIdTableFromVector({{2}, {2}}));
+
+  size_t counter = 0;
+  std::optional<Id> lastId;
+  for (const IdTable& table : tables) {
+    IndexImpl::countDistinct(lastId, counter, table);
+  }
+  EXPECT_EQ(counter, 2);
+}

test/ParallelExecutorTest.cpp

@@ -0,0 +1,82 @@
+//   Copyright 2026 The QLever Authors, in particular:
+//
+//  2026 Robin Textor-Falconi <[email protected]>, UFR
+//
+//  UFR = University of Freiburg, Chair of Algorithms and Data Structures
+
+#include <absl/strings/str_cat.h>
+#include <gtest/gtest.h>
+
+#include "backports/algorithm.h"
+#include "util/GTestHelpers.h"
+#include "util/ParallelExecutor.h"
+
+// _____________________________________________________________________________
+TEST(ParallelExecutor, noTasks) { ad_utility::runTasksInParallel({}); }
+
+// _____________________________________________________________________________
+TEST(ParallelExecutor, singleTask) {
+  bool executed = false;
+  std::vector<std::packaged_task<void()>> tasks;
+  tasks.push_back(std::packaged_task{[&executed]() { executed = true; }});
+  ad_utility::runTasksInParallel(std::move(tasks));
+  EXPECT_TRUE(executed);
+}
+
+// _____________________________________________________________________________
+TEST(ParallelExecutor, multipleTasks) {
+  constexpr size_t NUM_TASKS = 10;
+  std::array<bool, NUM_TASKS> executed;
+  ql::ranges::fill(executed, false);
+  std::vector<std::packaged_task<void()>> tasks;
+  for (size_t i = 0; i < NUM_TASKS; ++i) {
+    tasks.push_back(
+        std::packaged_task{[&executed, i]() { executed.at(i) = true; }});
+  }
+  ad_utility::runTasksInParallel(std::move(tasks));
+  for (size_t i = 0; i < NUM_TASKS; ++i) {
+    EXPECT_TRUE(executed.at(i));
+  }
+}
+
+// _____________________________________________________________________________
+TEST(ParallelExecutor, multipleTaskWithOneException) {
+  constexpr size_t NUM_TASKS = 10;
+  std::array<bool, NUM_TASKS> executed;
+  ql::ranges::fill(executed, false);
+  std::vector<std::packaged_task<void()>> tasks;
+  for (size_t i = 0; i < NUM_TASKS; ++i) {
+    tasks.push_back(std::packaged_task{[&executed, i]() {
+      executed.at(i) = true;
+      if (i == 5) {
+        throw std::runtime_error("Error");
+      }
+    }});
+  }
+  EXPECT_THROW(ad_utility::runTasksInParallel(std::move(tasks)),
+               std::runtime_error);
+  for (size_t i = 0; i < NUM_TASKS; ++i) {
+    EXPECT_TRUE(executed.at(i));
+  }
+}
+
+// _____________________________________________________________________________
+TEST(ParallelExecutor, multipleTaskWithOnlyExceptions) {
+  constexpr size_t NUM_TASKS = 10;
+  std::array<bool, NUM_TASKS> executed;
+  ql::ranges::fill(executed, false);
+  std::vector<std::packaged_task<void()>> tasks;
+  for (size_t i = 0; i < NUM_TASKS; ++i) {
+    tasks.push_back(std::packaged_task{[&executed, i]() {
+      executed.at(i) = true;
+      throw std::runtime_error(absl::StrCat("Error ", i));
+    }});
+  }
+  // Only the first error should be rethrown for simplicity.
+  AD_EXPECT_THROW_WITH_MESSAGE_AND_TYPE(
+      ad_utility::runTasksInParallel(std::move(tasks)),
+      ::testing::StrEq("Error 0"), std::runtime_error);
+  for (size_t i = 0; i < NUM_TASKS; ++i) {
+    EXPECT_TRUE(executed.at(i));
+  }
+}