Fix bugs in Connection::interrupt

src/include/main/client_context.h

@@ -15,6 +15,8 @@ struct ActiveQuery {
     explicit ActiveQuery();
     std::atomic<bool> interrupted;
     common::Timer timer;
+
+    void reset();
 };
 
 /**
@@ -33,12 +35,12 @@ class ClientContext {
 
     ~ClientContext() = default;
 
-    inline void interrupt() { activeQuery->interrupted = true; }
+    inline void interrupt() { activeQuery.interrupted = true; }
 
-    bool isInterrupted() const { return activeQuery->interrupted; }
+    bool isInterrupted() const { return activeQuery.interrupted; }
 
     inline bool isTimeOut() {
-        return isTimeOutEnabled() && activeQuery->timer.getElapsedTimeInMS() > timeoutInMS;
+        return isTimeOutEnabled() && activeQuery.timer.getElapsedTimeInMS() > timeoutInMS;
     }
 
     inline bool isTimeOutEnabled() const { return timeoutInMS != 0; }
@@ -48,8 +50,10 @@ class ClientContext {
     std::string getCurrentSetting(std::string optionName);
 
 private:
+    inline void resetActiveQuery() { activeQuery.reset(); }
+
     uint64_t numThreadsForExecution;
-    std::unique_ptr<ActiveQuery> activeQuery;
+    ActiveQuery activeQuery;
     uint64_t timeoutInMS;
 };
 

src/main/client_context.cpp

@@ -10,13 +10,18 @@ namespace main {
 
 ActiveQuery::ActiveQuery() : interrupted{false} {}
 
+void ActiveQuery::reset() {
+    interrupted = false;
+    timer = common::Timer();
+}
+
 ClientContext::ClientContext()
     : numThreadsForExecution{std::thread::hardware_concurrency()},
       timeoutInMS{common::ClientContextConstants::TIMEOUT_IN_MS} {}
 
 void ClientContext::startTimingIfEnabled() {
     if (isTimeOutEnabled()) {
-        activeQuery->timer.start();
+        activeQuery.timer.start();
     }
 }
 

src/main/connection.cpp

@@ -280,7 +280,7 @@ std::string Connection::getRelPropertyNames(const std::string& relTableName) {
 }
 
 void Connection::interrupt() {
-    clientContext->activeQuery->interrupted = true;
+    clientContext->interrupt();
 }
 
 void Connection::setQueryTimeOut(uint64_t timeoutInMS) {
@@ -328,7 +328,7 @@ void Connection::bindParametersNoLock(PreparedStatement* preparedStatement,
 
 std::unique_ptr<QueryResult> Connection::executeAndAutoCommitIfNecessaryNoLock(
     PreparedStatement* preparedStatement, uint32_t planIdx) {
-    clientContext->activeQuery = std::make_unique<ActiveQuery>();
+    clientContext->resetActiveQuery();
     clientContext->startTimingIfEnabled();
     auto mapper = PlanMapper(
         *database->storageManager, database->memoryManager.get(), database->catalog.get());

test/c_api/connection_test.cpp

@@ -208,19 +208,25 @@ TEST_F(CApiConnectionTest, QueryTimeout) {
 
 TEST_F(CApiConnectionTest, Interrupt) {
     auto connection = getConnection();
+    bool finished = false;
+
     // Interrupt the query after 100ms
-    std::thread t([&connection]() {
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
-        kuzu_connection_interrupt(connection);
+    // This may happen too early, so try again until the query function finishes.
+    std::thread t([&connection, &finished]() {
+        do {
+            std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            kuzu_connection_interrupt(connection);
+        } while (!finished);
     });
-    t.detach();
     auto result = kuzu_connection_query(
         connection, "MATCH (a:person)-[:knows*1..28]->(b:person) RETURN COUNT(*);");
+    finished = true;
     ASSERT_NE(result, nullptr);
     ASSERT_NE(result->_query_result, nullptr);
     auto resultCpp = static_cast<QueryResult*>(result->_query_result);
     ASSERT_NE(resultCpp, nullptr);
     ASSERT_FALSE(resultCpp->isSuccess());
     ASSERT_EQ(resultCpp->getErrorMessage(), "Interrupted.");
     kuzu_query_result_destroy(result);
+    t.join();
 }

test/runner/e2e_copy_transaction_test.cpp

@@ -80,7 +80,7 @@ class TinySnbCopyCSVTransactionTest : public EmptyDBTest {
         auto physicalPlan =
             mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlans[0].get(),
                 preparedStatement->statementResult->getColumns());
-        clientContext->activeQuery = std::make_unique<ActiveQuery>();
+        clientContext->resetActiveQuery();
         getQueryProcessor(*database)->execute(physicalPlan.get(), executionContext.get());
         auto tableID = catalog->getReadOnlyVersion()->getTableID("person");
         validateDatabaseStateBeforeCheckPointCopyNode(tableID);
@@ -159,7 +159,7 @@ class TinySnbCopyCSVTransactionTest : public EmptyDBTest {
         auto physicalPlan =
             mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlans[0].get(),
                 preparedStatement->statementResult->getColumns());
-        clientContext->activeQuery = std::make_unique<ActiveQuery>();
+        clientContext->resetActiveQuery();
         getQueryProcessor(*database)->execute(physicalPlan.get(), executionContext.get());
         auto tableID = catalog->getReadOnlyVersion()->getTableID("knows");
         validateDatabaseStateBeforeCheckPointCopyRel(tableID);

test/runner/e2e_ddl_test.cpp

@@ -277,7 +277,7 @@ class TinySnbDDLTest : public DBTest {
         auto physicalPlan =
             mapper.mapLogicalPlanToPhysical(preparedStatement->logicalPlans[0].get(),
                 preparedStatement->statementResult->getColumns());
-        executionContext->clientContext->activeQuery = std::make_unique<ActiveQuery>();
+        executionContext->clientContext->resetActiveQuery();
         getQueryProcessor(*database)->execute(physicalPlan.get(), executionContext.get());
     }