TUR-292: Versioning: count a commit's nodes, edges and dataparts when COMMIT builds them

storage/versioning/CommitBuilder.cpp

@@ -113,6 +113,8 @@ CommitResult<void> CommitBuilder::buildAllPending(JobSystem& jobsystem) {
 
     _builders.clear();
 
+    updateCommitStats();
+
     return {};
 }
 
@@ -123,20 +125,6 @@ CommitResult<std::unique_ptr<Commit>> CommitBuilder::build(JobSystem& jobsystem)
 
     _commit->history().journal().finalise();
 
-    const size_t numCommitDataParts = _commit->history().commitDataparts().size();
-    size_t numNodesAdded = 0;
-    size_t numEdgesAdded = 0;
-
-    for (size_t i = 0; i < numCommitDataParts; ++i) {
-        const DataPart* part = _commit->data().commitDataparts()[i].get();
-        numNodesAdded += part->getNodeContainerSize();
-        numEdgesAdded += part->getEdgeContainerSize();
-    }
-
-    _commit->_numDataParts = numCommitDataParts;
-    _commit->_numNodes = numNodesAdded;
-    _commit->_numEdges = numEdgesAdded;
-
     return std::move(_commit);
 }
 
@@ -212,6 +200,21 @@ void CommitBuilder::flushWriteBuffer([[maybe_unused]] JobSystem& jobsystem) {
     wb.setFlushed();
 }
 
+void CommitBuilder::updateCommitStats() {
+    const DataPartSpan commitDataparts = _commitData->commitDataparts();
+    size_t numNodesAdded = 0;
+    size_t numEdgesAdded = 0;
+
+    for (const WeakArc<DataPart>& part : commitDataparts) {
+        numNodesAdded += part->getNodeContainerSize();
+        numEdgesAdded += part->getEdgeContainerSize();
+    }
+
+    _commit->_numDataParts = commitDataparts.size();
+    _commit->_numNodes = numNodesAdded;
+    _commit->_numEdges = numEdgesAdded;
+}
+
 CommitBuilder::CommitBuilder(VersionController& controller, Change* change, const GraphView& view)
     : _controller(&controller),
     _change(change),

storage/versioning/CommitBuilder.h

@@ -114,5 +114,6 @@ class CommitBuilder {
 
     void initialize(const Commit* prevCommit);
     void initializeMerge(const Commit* prevCommit);
+    void updateCommitStats();
 };
 }

test/query/ir/WriteQueryTest.cpp

@@ -52,28 +52,26 @@ void WriteQueryTest::submit(const ChangeID& changeID) {
 }
 
 QueryStatus WriteQueryTest::runQuery(std::string_view query, NLOutputSink* sink) {
+    return runQuery(query, ChangeID::head(), sink);
+}
+
+QueryStatus WriteQueryTest::runQuery(std::string_view query,
+                                     const ChangeID& changeID,
+                                     NLOutputSink* sink) {
     QueryStatus status;
     _interpreter->execute(status,
                           query,
                           _graphName,
                           CommitHash::head(),
-                          ChangeID::head(),
+                          changeID,
                           sink);
 
     return status;
 }
 
 QueryStatus WriteQueryTest::runWrite(std::string_view query, const ChangeID& changeID) {
     NullSink sink;
-    QueryStatus status;
-    _interpreter->execute(status,
-                          query,
-                          _graphName,
-                          CommitHash::head(),
-                          changeID,
-                          &sink);
-
-    return status;
+    return runQuery(query, changeID, &sink);
 }
 
 void WriteQueryTest::applyWrite(std::string_view query) {

test/query/ir/WriteQueryTest.h

@@ -35,8 +35,9 @@ class WriteQueryTest : public TuringTest {
     void openChange(db::ChangeID& changeID);
     void submit(const db::ChangeID& changeID);
 
-    // A reading query at the graph's head, and a writing one in @param changeID
+    // A query at the graph's head or in @param changeID, and a writing one whose rows are dropped
     db::QueryStatus runQuery(std::string_view query, db::NLOutputSink* sink);
+    db::QueryStatus runQuery(std::string_view query, const db::ChangeID& changeID, db::NLOutputSink* sink);
     db::QueryStatus runWrite(std::string_view query, const db::ChangeID& changeID);
 
     // Runs a writing query in its own change and submits it, so a following read sees it

test/query/ir/write/CMakeLists.txt

@@ -93,3 +93,6 @@ add_turing_test(test_query_ir_cell_write_to_existing_property CellWriteToExistin
 
 # One write giving a property with no type two types.
 add_turing_test(test_query_ir_new_property_two_types NewPropertyTwoTypesTest.cpp)
+
+# The nodes, edges and dataparts db.history reports for a commit COMMIT closes in a change.
+add_turing_test(test_query_ir_change_commit_history ChangeCommitHistoryTest.cpp)

test/query/ir/write/ChangeCommitHistoryTest.cpp

@@ -0,0 +1,246 @@
+#include <gtest/gtest.h>
+
+#include <string>
+#include <string_view>
+
+#include "QueryStatus.h"
+
+#include "versioning/ChangeID.h"
+
+#include "IRTestRows.h"
+#include "WriteQueryTest.h"
+
+using namespace db;
+using namespace turing::test;
+
+// db.history reports, for each commit, the nodes, edges and dataparts that commit added.
+// A commit that COMMIT closes inside a change reports them before CHANGE SUBMIT, with the
+// values it reports after it. The open tip has added nothing, so it reports 0 / 0 / 0.
+class ChangeCommitHistoryTest : public WriteQueryTest {
+protected:
+    static constexpr std::string_view HISTORY_QUERY = "CALL db.history() YIELD nodeCount, edgeCount, partCount "
+                                                      "RETURN nodeCount, edgeCount, partCount";
+
+    void write(std::string_view query, const ChangeID& changeID) {
+        const QueryStatus status = runWrite(query, changeID);
+        ASSERT_TRUE(status.isOk()) << "query: " << query << "\nerror: " << status.getError();
+    }
+
+    void queryRows(std::string_view query, const ChangeID& changeID, Rows& rows) {
+        RowSink sink;
+        const QueryStatus status = runQuery(query, changeID, &sink);
+        ASSERT_TRUE(status.isOk()) << "query: " << query << "\nerror: " << status.getError();
+
+        rows = sink.rows();
+    }
+
+    void historyInChange(const ChangeID& changeID, Rows& rows) {
+        queryRows(HISTORY_QUERY, changeID, rows);
+    }
+
+    void historyAtHead(Rows& rows) {
+        queryRows(HISTORY_QUERY, ChangeID::head(), rows);
+    }
+
+    // The history head-first: @param changeRows for the commits the change made, then the
+    // history of the graph the change was opened on
+    void expectHistory(const Rows& actual, const Rows& changeRows, const Rows& baseRows) {
+        Rows expected = changeRows;
+        expected.insert(expected.end(), baseRows.begin(), baseRows.end());
+
+        std::string actualText;
+        describeRows(actual, actualText);
+
+        EXPECT_EQ(actual, expected) << "got:\n" << actualText;
+    }
+};
+
+TEST_F(ChangeCommitHistoryTest, committedCreateReportsItsNodesEdgesAndPart) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+    write("COMMIT", changeID);
+
+    Rows rows;
+    historyInChange(changeID, rows);
+
+    expectHistory(rows, {{"0", "0", "0"}, {"2", "1", "1"}}, baseRows);
+}
+
+TEST_F(ChangeCommitHistoryTest, committedCreateBetweenExistingNodesCountsOnlyWhatItAdded) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("MATCH (r:Person {name: 'Remy'}), (a:Person {name: 'Adam'}) "
+          "CREATE (r)-[:PARTY_TO]->(c:Crime {id: 'x'})-[:OCCURRED_AT]->(a)",
+          changeID);
+    write("COMMIT", changeID);
+
+    Rows rows;
+    historyInChange(changeID, rows);
+
+    expectHistory(rows, {{"0", "0", "0"}, {"1", "2", "1"}}, baseRows);
+}
+
+TEST_F(ChangeCommitHistoryTest, everyCommitOfAChangeReportsItsOwnCounts) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+    write("COMMIT", changeID);
+
+    write("CREATE (:Person {name: 'Z'}), (:Person {name: 'W'}), (:Person {name: 'V'})", changeID);
+    write("COMMIT", changeID);
+
+    Rows rows;
+    historyInChange(changeID, rows);
+
+    expectHistory(rows, {{"0", "0", "0"}, {"3", "0", "1"}, {"2", "1", "1"}}, baseRows);
+}
+
+TEST_F(ChangeCommitHistoryTest, oneCommitOfSeveralQueriesReportsTheirSum) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+    write("CREATE (:Person {name: 'Z'})", changeID);
+    write("COMMIT", changeID);
+
+    Rows rows;
+    historyInChange(changeID, rows);
+
+    expectHistory(rows, {{"0", "0", "0"}, {"3", "1", "1"}}, baseRows);
+}
+
+TEST_F(ChangeCommitHistoryTest, submitKeepsTheCountsTheChangeReported) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+    write("COMMIT", changeID);
+
+    write("CREATE (:Person {name: 'Z'})", changeID);
+    write("COMMIT", changeID);
+
+    Rows changeRows;
+    historyInChange(changeID, changeRows);
+
+    submit(changeID);
+
+    Rows headRows;
+    historyAtHead(headRows);
+
+    const Rows expectedChangeRows {{"0", "0", "0"}, {"1", "0", "1"}, {"2", "1", "1"}};
+    expectHistory(changeRows, expectedChangeRows, baseRows);
+    expectHistory(headRows, expectedChangeRows, baseRows);
+}
+
+// Main moves while the change is open, so the submit rebases the change's commits. The
+// rebase must not change what each commit added.
+TEST_F(ChangeCommitHistoryTest, rebaseOnSubmitKeepsTheCounts) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+    write("COMMIT", changeID);
+
+    Rows changeRows;
+    historyInChange(changeID, changeRows);
+    expectHistory(changeRows, {{"0", "0", "0"}, {"2", "1", "1"}}, baseRows);
+
+    applyWrite("CREATE (:City {name: 'Paris'})");
+
+    Rows mainRows;
+    historyAtHead(mainRows);
+
+    submit(changeID);
+
+    Rows headRows;
+    historyAtHead(headRows);
+
+    expectHistory(headRows, {{"0", "0", "0"}, {"2", "1", "1"}}, mainRows);
+}
+
+// A CREATE and a SET on an existing node flush into two dataparts of one commit
+TEST_F(ChangeCommitHistoryTest, commitWithCreatesAndUpdatesCountsBothParts) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (:Person {name: 'Z'})", changeID);
+    write("MATCH (n:Person {name: 'Remy'}) SET n.age = 1", changeID);
+    write("COMMIT", changeID);
+
+    Rows changeRows;
+    historyInChange(changeID, changeRows);
+
+    expectHistory(changeRows, {{"0", "0", "0"}, {"1", "0", "2"}}, baseRows);
+
+    submit(changeID);
+
+    Rows headRows;
+    historyAtHead(headRows);
+
+    EXPECT_EQ(changeRows, headRows);
+}
+
+TEST_F(ChangeCommitHistoryTest, describeCommitReportsACommittedCommit) {
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+    write("COMMIT", changeID);
+
+    Rows hashRows;
+    queryRows("CALL db.history() YIELD commit RETURN commit", changeID, hashRows);
+    ASSERT_GE(hashRows.size(), 2);
+
+    const std::string describeQuery = "CALL db.describeCommit('" + hashRows[1][0] + "') "
+                                      "YIELD nodeCount, edgeCount, partCount "
+                                      "RETURN nodeCount, edgeCount, partCount";
+    Rows rows;
+    queryRows(describeQuery, changeID, rows);
+
+    EXPECT_EQ(rows, (Rows {{"2", "1", "1"}}));
+}
+
+// The open tip has added nothing, even with writes staged in it before a COMMIT
+TEST_F(ChangeCommitHistoryTest, openTipReportsNothingBeforeCommit) {
+    Rows baseRows;
+    historyAtHead(baseRows);
+
+    ChangeID changeID;
+    openChange(changeID);
+
+    write("CREATE (a:Person {name: 'X'})-[:KNOWS]->(b:Person {name: 'Y'})", changeID);
+
+    Rows rows;
+    historyInChange(changeID, rows);
+
+    expectHistory(rows, {{"0", "0", "0"}}, baseRows);
+}
+
+int main(int argc, char** argv) {
+    return turing::test::turingTestMain(argc, argv);
+}