Add test for operation failure propagation

test/JoinTest.cpp

@@ -28,6 +28,7 @@
 #include "global/RuntimeParameters.h"
 #include "util/Forward.h"
 #include "util/IndexTestHelpers.h"
+#include "util/OperationTestHelpers.h"
 #include "util/Random.h"
 #include "util/SourceLocation.h"
 
@@ -672,3 +673,30 @@ TEST(JoinTest, joinLazyAndNonLazyOperationWithAndWithoutUndefValues) {
   performJoin(createIdTableOfSizeWithValue(Join::CHUNK_SIZE, I(1)),
               std::move(rightTables), expected7, false);
 }
+
+// _____________________________________________________________________________
+TEST(JoinTest, errorInSeparateThreadIsPropagatedCorrectly) {
+  auto qec = ad_utility::testing::getQec();
+  RuntimeParameters().set<"lazy-index-scan-max-size-materialization">(0);
+  absl::Cleanup cleanup{[]() {
+    // Reset back to original value to not influence other tests.
+    RuntimeParameters().set<"lazy-index-scan-max-size-materialization">(
+        1'000'000);
+  }};
+  auto leftTree =
+      ad_utility::makeExecutionTree<AlwaysFailOperation>(qec, Variable{"?s"});
+  auto rightTree = ad_utility::makeExecutionTree<ValuesForTesting>(
+      qec, makeIdTableFromVector({{I(1)}}), Vars{Variable{"?s"}}, false,
+      std::vector<ColumnIndex>{0});
+  VariableToColumnMap expectedVariables{
+      {Variable{"?s"}, makeAlwaysDefinedColumn(0)}};
+  Join join{qec, leftTree, rightTree, 0, 0};
+
+  auto result = join.getResult(false, ComputationMode::LAZY_IF_SUPPORTED);
+  ASSERT_FALSE(result->isFullyMaterialized());
+
+  auto& idTables = result->idTables();
+  AD_EXPECT_THROW_WITH_MESSAGE_AND_TYPE(idTables.begin(),
+                                        testing::StrEq("AlwaysFailOperation"),
+                                        std::runtime_error);
+}

test/util/OperationTestHelpers.h

@@ -87,21 +87,37 @@ class ShallowParentOperation : public Operation {
 
 // Operation that will throw on `computeResult` for testing.
 class AlwaysFailOperation : public Operation {
+  static std::atomic_uint32_t cacheCounter;
+
+  std::optional<Variable> variable_ = std::nullopt;
+
   std::vector<QueryExecutionTree*> getChildren() override { return {}; }
-  string getCacheKeyImpl() const override { AD_FAIL(); }
+  string getCacheKeyImpl() const override {
+    // Every operation gets a unique cache key
+    return absl::StrCat("AlwaysFailCacheKey_", cacheCounter++);
+  }
   string getDescriptor() const override {
     return "AlwaysFailOperationDescriptor";
   }
-  size_t getResultWidth() const override { return 0; }
+  size_t getResultWidth() const override { return 1; }
   size_t getCostEstimate() override { return 0; }
   uint64_t getSizeEstimateBeforeLimit() override { return 0; }
   float getMultiplicity([[maybe_unused]] size_t) override { return 0; }
   bool knownEmptyResult() override { return false; }
-  vector<ColumnIndex> resultSortedOn() const override { return {}; }
-  VariableToColumnMap computeVariableToColumnMap() const override { return {}; }
+  vector<ColumnIndex> resultSortedOn() const override { return {0}; }
+  VariableToColumnMap computeVariableToColumnMap() const override {
+    if (!variable_.has_value()) {
+      return {};
+    }
+    return {{variable_.value(),
+             ColumnIndexAndTypeInfo{
+                 0, ColumnIndexAndTypeInfo::UndefStatus::AlwaysDefined}}};
+  }
 
  public:
   using Operation::Operation;
+  AlwaysFailOperation(QueryExecutionContext* qec, Variable variable)
+      : Operation{qec}, variable_{std::move(variable)} {}
   ProtoResult computeResult(bool requestLaziness) override {
     if (!requestLaziness) {
       throw std::runtime_error{"AlwaysFailOperation"};
@@ -115,6 +131,8 @@ class AlwaysFailOperation : public Operation {
   }
 };
 
+std::atomic_uint32_t AlwaysFailOperation::cacheCounter = 0;
+
 // Lazy operation that will yield a result with a custom generator you can
 // provide via the constructor.
 class CustomGeneratorOperation : public Operation {