Merge pull request #2012 from kuzudb/transaction-statement

[WIP] Transaction statement

src/antlr4/Cypher.g4

@@ -66,7 +66,8 @@ kU_DDL
         | kU_CreateRelTableGroup
         | kU_CreateRdfGraph
         | kU_DropTable
-        | kU_AlterTable;
+        | kU_AlterTable
+        ;
 
 kU_CreateNodeTable
     : CREATE SP NODE SP TABLE SP oC_SchemaName SP? '(' SP? kU_PropertyDefinitions SP? ( ',' SP? kU_CreateNodeConstraint ) SP? ')' ;
@@ -173,7 +174,32 @@ oC_Statement
         | kU_CopyFromCSV
         | kU_CopyTO
         | kU_StandaloneCall
-        | kU_CreateMacro ;
+        | kU_CreateMacro
+        | kU_Transaction ;
+
+kU_Transaction
+    : BEGIN SP READ SP TRANSACTION
+        | BEGIN SP WRITE SP TRANSACTION
+        | COMMIT
+        | COMMIT_SKIP_CHECKPOINT
+        | ROLLBACK
+        | ROLLBACK_SKIP_CHECKPOINT;
+
+BEGIN : ( 'B' | 'b' ) ( 'E' | 'e' ) ( 'G' | 'g' ) ( 'I' | 'i' ) ( 'N' | 'n' ) ;
+
+TRANSACTION : ( 'T' | 't' ) ( 'R' | 'r' ) ( 'A' | 'a' ) ( 'N' | 'n' ) ( 'S' | 's' ) ( 'A' | 'a' ) ( 'C' | 'c' ) ( 'T' | 't' ) ( 'I' | 'i' ) ( 'O' | 'o' ) ( 'N' | 'n' ) ;
+
+READ : ( 'R' | 'r' ) ( 'E' | 'e' ) ( 'A' | 'a' ) ( 'D' | 'd' ) ;
+
+WRITE : ( 'W' | 'w' ) ( 'R' | 'r' ) ( 'I' | 'i' ) ( 'T' | 't' ) ( 'E' | 'e' ) ;
+
+COMMIT : ( 'C' | 'c' ) ( 'O' | 'o' ) ( 'M' | 'm' ) ( 'M' | 'm' ) ( 'I' | 'i' ) ( 'T' | 't' ) ;
+
+COMMIT_SKIP_CHECKPOINT : ( 'C' | 'c' ) ( 'O' | 'o' ) ( 'M' | 'm' ) ( 'M' | 'm' ) ( 'I' | 'i' ) ( 'T' | 't' ) '_' ( 'S' | 's' ) ( 'K' | 'k' ) ( 'I' | 'i' ) ( 'P' | 'p' ) '_' ( 'C' | 'c' ) ( 'H' | 'h' ) ( 'E' | 'e' ) ( 'C' | 'c' ) ( 'K' | 'k' ) ( 'P' | 'p' ) ( 'O' | 'o' ) ( 'I' | 'i' ) ( 'N' | 'n' ) ( 'T' | 't' ) ;
+
+ROLLBACK : ( 'R' | 'r' ) ( 'O' | 'o' ) ( 'L' | 'l' ) ( 'L' | 'l' ) ( 'B' | 'b' ) ( 'A' | 'a' ) ( 'C' | 'c' ) ( 'K' | 'k' ) ;
+
+ROLLBACK_SKIP_CHECKPOINT: ( 'R' | 'r' ) ( 'O' | 'o' ) ( 'L' | 'l' ) ( 'L' | 'l' ) ( 'B' | 'b' ) ( 'A' | 'a' ) ( 'C' | 'c' ) ( 'K' | 'k' ) '_' ( 'S' | 's' ) ( 'K' | 'k' ) ( 'I' | 'i' ) ( 'P' | 'p' ) '_' ( 'C' | 'c' ) ( 'H' | 'h' ) ( 'E' | 'e' ) ( 'C' | 'c' ) ( 'K' | 'k' ) ( 'P' | 'p' ) ( 'O' | 'o' ) ( 'I' | 'i' ) ( 'N' | 'n' ) ( 'T' | 't' ) ;
 
 oC_Query
     : oC_RegularQuery ;

src/binder/bind/CMakeLists.txt

@@ -2,6 +2,7 @@ add_library(
         kuzu_binder_bind
         OBJECT
         bind_standalone_call.cpp
+        bind_transaction.cpp
         bind_create_macro.cpp
         bind_copy.cpp
         bind_ddl.cpp

src/binder/bind/bind_transaction.cpp

@@ -0,0 +1,16 @@
+#include "binder/binder.h"
+#include "binder/bound_transaction_statement.h"
+#include "parser/transaction_statement.h"
+
+using namespace kuzu::parser;
+
+namespace kuzu {
+namespace binder {
+
+std::unique_ptr<BoundStatement> Binder::bindTransaction(const Statement& statement) {
+    auto transactionStatement = reinterpret_cast<const TransactionStatement&>(statement);
+    return std::make_unique<BoundTransactionStatement>(transactionStatement.getTransactionAction());
+}
+
+} // namespace binder
+} // namespace kuzu

src/binder/binder.cpp

@@ -50,6 +50,9 @@ std::unique_ptr<BoundStatement> Binder::bind(const Statement& statement) {
     case StatementType::CREATE_MACRO: {
         boundStatement = bindCreateMacro(statement);
     } break;
+    case StatementType::TRANSACTION: {
+        boundStatement = bindTransaction(statement);
+    } break;
     default:
         throw NotImplementedException("Binder::bind");
     }

src/binder/bound_statement_visitor.cpp

@@ -43,6 +43,9 @@ void BoundStatementVisitor::visit(const kuzu::binder::BoundStatement& statement)
     case StatementType::CREATE_MACRO: {
         visitCreateMacro(statement);
     } break;
+    case StatementType::TRANSACTION: {
+        visitTransaction(statement);
+    } break;
     default:
         throw NotImplementedException("BoundStatementVisitor::visit");
     }

src/binder/visitor/statement_read_write_analyzer.cpp

@@ -9,7 +9,9 @@ bool StatementReadWriteAnalyzer::isReadOnly(const kuzu::binder::BoundStatement&
 }
 
 void StatementReadWriteAnalyzer::visitQueryPart(const NormalizedQueryPart& queryPart) {
-    readOnly = !queryPart.hasUpdatingClause();
+    if (queryPart.hasUpdatingClause()) {
+        readOnly = false;
+    }
 }
 
 } // namespace binder

src/c_api/connection.cpp

@@ -33,22 +33,6 @@ void kuzu_connection_destroy(kuzu_connection* connection) {
     free(connection);
 }
 
-void kuzu_connection_begin_read_only_transaction(kuzu_connection* connection) {
-    static_cast<Connection*>(connection->_connection)->beginReadOnlyTransaction();
-}
-
-void kuzu_connection_begin_write_transaction(kuzu_connection* connection) {
-    static_cast<Connection*>(connection->_connection)->beginWriteTransaction();
-}
-
-void kuzu_connection_commit(kuzu_connection* connection) {
-    static_cast<Connection*>(connection->_connection)->commit();
-}
-
-void kuzu_connection_rollback(kuzu_connection* connection) {
-    static_cast<Connection*>(connection->_connection)->rollback();
-}
-
 void kuzu_connection_set_max_num_thread_for_exec(
     kuzu_connection* connection, uint64_t num_threads) {
     static_cast<Connection*>(connection->_connection)->setMaxNumThreadForExec(num_threads);

src/catalog/catalog.cpp

@@ -3,7 +3,6 @@
 #include "catalog/rel_table_group_schema.h"
 #include "common/ser_deser.h"
 #include "common/string_utils.h"
-#include "storage/storage_utils.h"
 
 using namespace kuzu::common;
 using namespace kuzu::storage;

src/include/binder/binder.h

@@ -135,6 +135,9 @@ class Binder {
     /*** bind create macro ***/
     std::unique_ptr<BoundStatement> bindCreateMacro(const parser::Statement& statement);
 
+    /*** bind transaction ***/
+    std::unique_ptr<BoundStatement> bindTransaction(const parser::Statement& statement);
+
     /*** bind explain ***/
     std::unique_ptr<BoundStatement> bindExplain(const parser::Statement& statement);
 

src/include/binder/bound_statement_visitor.h

@@ -28,6 +28,7 @@ class BoundStatementVisitor {
     virtual void visitStandaloneCall(const BoundStatement& statement) {}
     virtual void visitExplain(const BoundStatement& statement);
     virtual void visitCreateMacro(const BoundStatement& statement) {}
+    virtual void visitTransaction(const BoundStatement& statement) {}
 
     void visitReadingClause(const BoundReadingClause& readingClause);
     virtual void visitMatch(const BoundReadingClause& readingClause) {}

src/include/binder/bound_transaction_statement.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "bound_statement.h"
+#include "transaction/transaction_action.h"
+
+namespace kuzu {
+namespace binder {
+
+class BoundTransactionStatement : public BoundStatement {
+public:
+    explicit BoundTransactionStatement(transaction::TransactionAction transactionAction)
+        : BoundStatement{common::StatementType::TRANSACTION,
+              BoundStatementResult::createEmptyResult()},
+          transactionAction{transactionAction} {}
+
+    inline transaction::TransactionAction getTransactionAction() const { return transactionAction; }
+
+private:
+    transaction::TransactionAction transactionAction;
+};
+
+} // namespace binder
+} // namespace kuzu

src/include/binder/visitor/statement_read_write_analyzer.h

@@ -7,11 +7,19 @@ namespace binder {
 
 class StatementReadWriteAnalyzer : public BoundStatementVisitor {
 public:
-    StatementReadWriteAnalyzer() : BoundStatementVisitor(), readOnly{false} {}
+    StatementReadWriteAnalyzer() : BoundStatementVisitor(), readOnly{true} {}
 
     bool isReadOnly(const BoundStatement& statement);
 
 private:
+    void visitCreateTable(const BoundStatement& statement) override { readOnly = false; }
+    void visitDropTable(const BoundStatement& statement) override { readOnly = false; }
+    void visitRenameTable(const BoundStatement& statement) override { readOnly = false; }
+    void visitAddProperty(const BoundStatement& statement) override { readOnly = false; }
+    void visitDropProperty(const BoundStatement& statement) override { readOnly = false; }
+    void visitRenameProperty(const BoundStatement& statement) override { readOnly = false; }
+    void visitCopy(const BoundStatement& statement) override { readOnly = false; }
+    void visitCreateMacro(const BoundStatement& statement) override { readOnly = false; }
     void visitQueryPart(const NormalizedQueryPart& queryPart) final;
 
 private:

src/include/c_api/kuzu.h

@@ -217,26 +217,6 @@ KUZU_C_API kuzu_connection* kuzu_connection_init(kuzu_database* database);
  * @param connection The connection instance to destroy.
  */
 KUZU_C_API void kuzu_connection_destroy(kuzu_connection* connection);
-/**
- * @brief Begins a read-only transaction in the given connection.
- * @param connection The connection instance to begin read-only transaction.
- */
-KUZU_C_API void kuzu_connection_begin_read_only_transaction(kuzu_connection* connection);
-/**
- * @brief Begins a write transaction in the given connection.
- * @param connection The connection instance to begin write transaction.
- */
-KUZU_C_API void kuzu_connection_begin_write_transaction(kuzu_connection* connection);
-/**
- * @brief Commits the current transaction.
- * @param connection The connection instance to commit transaction.
- */
-KUZU_C_API void kuzu_connection_commit(kuzu_connection* connection);
-/**
- * @brief Rollbacks the current transaction.
- * @param connection The connection instance to rollback transaction.
- */
-KUZU_C_API void kuzu_connection_rollback(kuzu_connection* connection);
 /**
  * @brief Sets the maximum number of threads to use for executing queries.
  * @param connection The connection instance to set max number of threads for execution.

src/include/common/statement_type.h

@@ -18,6 +18,7 @@ enum class StatementType : uint8_t {
     STANDALONE_CALL = 21,
     EXPLAIN = 22,
     CREATE_MACRO = 23,
+    TRANSACTION = 30,
 };
 
 struct StatementTypeUtils {

src/include/main/client_context.h

@@ -15,6 +15,7 @@ class Binder;
 }
 
 namespace main {
+class Database;
 
 struct ActiveQuery {
     explicit ActiveQuery();
@@ -38,9 +39,9 @@ class ClientContext {
     friend class VarLengthExtendMaxDepthSetting;
 
 public:
-    explicit ClientContext();
+    explicit ClientContext(Database* database);
 
-    ~ClientContext() = default;
+    ~ClientContext();
 
     inline void interrupt() { activeQuery.interrupted = true; }
 
@@ -56,13 +57,17 @@ class ClientContext {
 
     std::string getCurrentSetting(std::string optionName);
 
+    transaction::Transaction* getActiveTransaction() const;
+    transaction::TransactionContext* getTransactionContext() const;
+
 private:
     inline void resetActiveQuery() { activeQuery.reset(); }
 
     uint64_t numThreadsForExecution;
     ActiveQuery activeQuery;
     uint64_t timeoutInMS;
     uint32_t varLengthExtendMaxDepth;
+    std::unique_ptr<transaction::TransactionContext> transactionContext;
 };
 
 } // namespace main

src/include/main/connection.h

@@ -23,25 +23,6 @@ class Connection {
     friend class kuzu::benchmark::Benchmark;
     friend class kuzu::testing::TinySnbDDLTest;
 
-public:
-    /**
-     * If the connection is in AUTO_COMMIT mode any query over the connection will be wrapped around
-     * a transaction and committed (even if the query is READ_ONLY).
-     * If the connection is in MANUAL transaction mode, which happens only if an application
-     * manually begins a transaction (see below), then an application has to manually commit or
-     * rollback the transaction by calling commit() or rollback().
-     *
-     * AUTO_COMMIT is the default mode when a Connection is created. If an application calls
-     * begin[ReadOnly/Write]Transaction at any point, the mode switches to MANUAL. This creates
-     * an "active transaction" in the connection. When a connection is in MANUAL mode and the
-     * active transaction is rolled back or committed, then the active transaction is removed (so
-     * the connection no longer has an active transaction) and the mode automatically switches
-     * back to AUTO_COMMIT.
-     * Note: When a Connection object is deconstructed, if the connection has an active (manual)
-     * transaction, then the active transaction is rolled back.
-     */
-    enum class ConnectionTransactionMode : uint8_t { AUTO_COMMIT = 0, MANUAL = 1 };
-
 public:
     /**
      * @brief Creates a connection to the database.
@@ -180,32 +161,9 @@ class Connection {
     }
 
 protected:
-    ConnectionTransactionMode getTransactionMode();
-    void setTransactionModeNoLock(ConnectionTransactionMode newTransactionMode);
-
     std::unique_ptr<QueryResult> query(const std::string& query, const std::string& encodedJoin);
-    // Note: This is only added for testing recovery algorithms in unit tests. Do not use
-    // this otherwise.
-    void commitButSkipCheckpointingForTestingRecovery();
-    // Note: This is only added for testing recovery algorithms in unit tests. Do not use
-    // this otherwise.
-    void rollbackButSkipCheckpointingForTestingRecovery();
-    // Note: This is only added for testing recovery algorithms in unit tests. Do not use
-    // this otherwise.
-    transaction::Transaction* getActiveTransaction();
-    // used in API test
-
-    uint64_t getActiveTransactionID();
-    bool hasActiveTransaction();
-    void commitNoLock();
-    void rollbackIfNecessaryNoLock();
 
-    void beginTransactionNoLock(transaction::TransactionType type);
-
-    void commitOrRollbackNoLock(
-        transaction::TransactionAction action, bool skipCheckpointForTesting = false);
-
-    std::unique_ptr<QueryResult> queryResultWithError(std::string& errMsg);
+    std::unique_ptr<QueryResult> queryResultWithError(const std::string& errMsg);
 
     std::unique_ptr<PreparedStatement> prepareNoLock(const std::string& query,
         bool enumerateAllPlans = false, std::string joinOrder = std::string{});
@@ -226,21 +184,11 @@ class Connection {
     std::unique_ptr<QueryResult> executeAndAutoCommitIfNecessaryNoLock(
         PreparedStatement* preparedStatement, uint32_t planIdx = 0u);
 
-    void beginTransactionIfAutoCommit(PreparedStatement* preparedStatement);
-
-private:
-    inline std::unique_ptr<QueryResult> getQueryResultWithError(std::string exceptionMessage) {
-        rollbackIfNecessaryNoLock();
-        return queryResultWithError(exceptionMessage);
-    }
-
     void addScalarFunction(std::string name, function::vector_function_definitions definitions);
 
 protected:
     Database* database;
     std::unique_ptr<ClientContext> clientContext;
-    std::unique_ptr<transaction::Transaction> activeTransaction;
-    ConnectionTransactionMode transactionMode;
     std::mutex mtx;
 };
 

src/include/main/database.h

@@ -36,6 +36,7 @@ class Database {
     friend class Connection;
     friend class StorageDriver;
     friend class kuzu::testing::BaseGraphTest;
+    friend class transaction::TransactionContext;
 
 public:
     /**

src/include/main/kuzu_fwd.h

@@ -52,9 +52,8 @@ class PhysicalPlan;
 
 namespace transaction {
 class Transaction;
-enum class TransactionType : uint8_t;
-enum class TransactionAction : uint8_t;
 class TransactionManager;
+class TransactionContext;
 } // namespace transaction
 
 } // namespace kuzu

src/include/main/prepared_statement.h

@@ -30,6 +30,7 @@ class PreparedStatement {
      * @return the prepared statement is allowed to be part of an active transaction.
      */
     KUZU_API bool allowActiveTransaction() const;
+    bool isTransactionStatement() const;
     /**
      * @return the query is prepared successfully or not.
      */

src/include/parser/explain_statement.h

@@ -8,8 +8,7 @@ namespace parser {
 
 class ExplainStatement : public Statement {
 public:
-    explicit ExplainStatement(
-        std::unique_ptr<Statement> statementToExplain, common::ExplainType explainType)
+    ExplainStatement(std::unique_ptr<Statement> statementToExplain, common::ExplainType explainType)
         : Statement{common::StatementType::EXPLAIN},
           statementToExplain{std::move(statementToExplain)}, explainType{explainType} {}