[DE-771] Issue with running queries in stream transaction

[jurisitsm]

Hi! An application my team is building, uses arango database to process large chunks incoming data periodically. For each individual batch of incoming data there is a large number of different db operations that have to executed (insertion, deletion, queries, updates etc.) The current implementation has some performance issues, therefore we decided to rewrite the code responsible for executing these db operations to use arangodb's stream transaction. The problem is, it seems that querying data, that has been added in the same transaction that it is being queried in, does not always return the legitimate results. I am currently experiencing this with edge collections.

We start the transaction the following way:

 var transaction = db.beginStreamTransaction(
        new StreamTransactionOptions().writeCollections(collections));    //array with the names of collection used
    try {
      //process data
      db.commitStreamTransaction(transaction.getId());
    } catch (Exception e) {
      db.abortStreamTransaction(transaction.getId());
    }

During this processing, if I try to query nodes that have been added in this same transaction, I can successfully achieve it like so:

  private static final String GET_NODES_QUERY = """
      FOR nodeEntity IN nodeEntities
          RETURN nodeEntity
      """;

    public List<NodeEntity> getNodeEntitiesInTransaction(String transactionId) {
    return arangoOperations.query(GET_NODES_QUERY, createQueryOptions(transactionId), NodeEntity.class).asListRemaining();
  }

    private AqlQueryOptions createQueryOptions(String transactionId) {
    var options = new AqlQueryOptions();
    options.streamTransactionId(transactionId);
    return options;
  }

But if I try to do the exact same logic, but instead of a collection annotated Document, it is one annotated Edge, the query always returns an empty response, even though I know, several instances of the given edge class have already been added. Apart from debugging I also know this, because after the transaction with given id is committed, the instances of the edge classes become immediately available. I would appreciate help with this issue. Additionally, I have another question, that is, whether the query method of the ArangoOperations class can only execute non-modifying queries, or any AQL statement provided as string (deletion for instance).

Thank you in advance.

[rashtao]

What you describe should just work fine. In order to debug it further, you can enable debug level logging for requests and response (logger com.arangodb.internal.net.Communication) and verify the presence of x-arango-trx-id headers in the requests sent within the transaction.

In case this is a bug in ArangoDB Spring Data or in the underlying java driver, please provide a code snippet to reproduce the problem.

ArangoOperations#query() can be used to execute any AQL statement.

[jurisitsm]

Hi! I think I might have found my mistake, I attempted to insert an edge entity in the following way: arangoOperations.insert(edgeInstance, new DocumentCreateOptions().streamTransactionId(transaction.getId())); And this does not work (if it should, I can provide you the code snippet, if needed). But now I see there is also a class called 'EdgeCreateOptions'. Am I correct in assuming, that this options class should be used for inserting edges? Remaining problem is only, I can't find any arangoOperations method that accepts this edge class as parameter, and I have trouble finding it in the documentation also. Can you provide an example for the usage of EdgeCreateOptions, for inserting edges into the db?

[rashtao]

I don't think that arangoOperations.insert(edgeInstance, new DocumentCreateOptions().streamTransactionId(transaction.getId())); is a mistake. You can double check whether it sends the transaction headers in the communication logs as mentioned above.

Note that ArangoOperations#insert(T, DocumentCreateOptions) inserts the document using the document API, while EdgeCreateOptions is used in com.arangodb.ArangoEdgeCollection#insertEdge(Object, EdgeCreateOptions) and uses the graph API.

The latter is not exposed from Spring Data ArangoDB API, but only from the underlying Java driver.

[jurisitsm]

Checking the logs you recommended yields the following results:

1. Logs of the insertion of the edge class, shows that the x-arango-trx-id header is in the request, alongside the correct transaction id: 2024-01-25 11:38:05,254 DEBUG [main] c.a.i.v.VstCommunicationSync: Send Message (id=1582, head=[1,1,"codee_backend",2,"/_api/document/issueReporters",{},{"x-arango-trx-id":"1790","x-arango-driver":"JavaDriver/6.22.0 (JVM/17)","content-type":"application/x-velocypack","accept":"application/x-velocypack"}], body={"_from":"issues/94f2d912342a6b1a541a968d2f73cd6aaddc0b9cdc066bba08fd643ac826d9ef","_to":"contributors/5f21a1996c0cf3c60b9edef2a94218220f687629536b985a0e8721d52dc4f80b","_key":"2afe3cf18cba1abeaf0f9814e863eac05b1b841aed63e17ee2f6bc63148e6ffd","accountId":5,"privateElement":true,"label":"reportedBy","_class":"com.frontendart.codee.backend.entity.domain.edge.IssueReporter"}) -
2. There is also a log, showing that the previous message was received: 2024-01-25 11:38:05,255 DEBUG [pool-6-thread-1] c.a.i.v.internal.MessageStore: Received Message (id=1582, head=[1,2,202,{"Location":"/_db/codee_backend/_api/document/issueReporters/2afe3cf18cba1abeaf0f9814e863eac05b1b841aed63e17ee2f6bc63148e6ffd","Etag":"\"_hS-wu6a---\""}], body={"_id":"issueReporters/2afe3cf18cba1abeaf0f9814e863eac05b1b841aed63e17ee2f6bc63148e6ffd","_key":"2afe3cf18cba1abeaf0f9814e863eac05b1b841aed63e17ee2f6bc63148e6ffd","_rev":"_hS-wu6a---"}) -
3. Upon querying the edges, I can also see the correct transaction id in the request (it's a different run that is why the id is different from the previous two): 2024-01-25 12:23:25,702 DEBUG [main] c.a.i.v.VstCommunicationSync: Send Message (id=1633, head=[1,1,"codee_backend",2,"/_api/cursor",{},{"x-arango-trx-id":"1938","x-arango-driver":"JavaDriver/6.22.0 (JVM/17)","content-type":"application/x-velocypack","accept":"application/x-velocypack"}], body={"query":"FOR issueReporter IN issueReporters RETURN issueReporter","streamTransactionId":"1938"}) -
4. Finally later on in the logs, after the query of the edge entities, there is an exception thrown: 2024-01-25 12:23:25,812 DEBUG [main] com.arangodb.ArangoCollection: Response: 404, Error: 1202 - document not found - com.arangodb.ArangoDBException: Response: 404, Error: 1202 - document not found (I can also provide the full log, if that helps)

Here is the test that resulted in the logs above, it is simply adding two nodes and an edge between them in the same transaction:

@Autowired
  private ArangoOperations arangoOperations;

  @Test
  public void streamTransactionTest() {
    ArangoDB arangoDB = arangoOperations.driver();
    DbName dbName = DbName.of("database_name");
    ArangoDatabase db = arangoDB.db(dbName);
    var transaction = db.beginStreamTransaction(
            new StreamTransactionOptions().writeCollections("entityAs", "entityBs", "edgeEntities"));
    try {
      EntityA entityA = new EntityA("id", "name");    //Entity annotated @Document
      arangoOperations.insert(entityA, new DocumentCreateOptions().streamTransactionId(transaction.getId()));

      EntityB entityB = new EntityB("id", "name"); //Other entity annotated @Document
      arangoOperations.insert(entityB, new DocumentCreateOptions().streamTransactionId(transaction.getId()));

      EdgeEntity edgeEntity = new EdgeEntity(entityA, entityB); //Entity annotated @Edge
      arangoOperations.insert(edgeEntity, new DocumentCreateOptions().streamTransactionId(transaction.getId()));

      arangoOperations.query("FOR entityA IN entityAs RETURN entityA",
              new AqlQueryOptions().streamTransactionId(transaction.getId()), EntityA.class).asListRemaining();  //returns list of one instance of entityA

      arangoOperations.query("FOR edgeEntity IN edgeEntities RETURN edgeEntity",
              new AqlQueryOptions().streamTransactionId(transaction.getId()), EdgeEntity.class).asListRemaining(); //throws java.util.NoSuchElementException: No value present

      db.commitStreamTransaction(transaction.getId());
    } catch (Exception e) {
      db.abortStreamTransaction(transaction.getId());
    }
  }

Important to note that if you take out the line querying the edge class (in which case no exception is thrown), after the transaction gets committed, the edge class can be queried from the database normally. (for example through executing arangoOperations.findAll(EdgeEntity.class)). Which means that adding the edge in transaction seems to work, but querying it does not. There is also a similar issue, with probably the same cause, that if I execute a query, with AqlQueryOptions, with the transaction id, that modifies edges, the changes also do not materialize, probably, because the edges to be modified can not be found by a query in the first place, the same way as above. This is what the query looks like:

    Map<String, Object> bindVars = new HashMap<>();
    bindVars.put("node", node.getArangoID());
    bindVars.put("@edge", edgeClazz);
    arangoOperations
            .query("FOR v, e IN 0..1 ANY @node @@edge FILTER v.booleanProperty REMOVE e IN @@edge", 
                    bindVars,
                    new AqlQueryOptions().streamTransactionId(transactionId), 
                    edgeClazz);

Any suggestion as to how could this functionality be implemented would be highly appreciated.

[jurisitsm]

I've found another important part of the logs. Right after I query the edge classes I see this snippet in the logs: 2024-01-26 10:17:08,507 DEBUG [pool-8-thread-1] c.a.i.v.internal.MessageStore: Received Message (id=1559, head=[1,2,201,{}], body={"result":[{"_key":"9fe6d6ab289fdfdc0b9acd3d887637d1405a707df65fa2e329160086a37e055f","_id":"issueReporters/9fe6d6ab289fdfdc0b9acd3d887637d1405a707df65fa2e329160086a37e055f","_from":"issues/f2b2b9bf97cbf375b9ab9d8e8c94e4011d11e55e38918f9cf1082d190f631e5e","_to":"contributors/48fd4ee80d544c831fdbf0c8872ca4784d95600d56cffdc18248a93b91db94f7","_rev":"_hSSM85i---","accountId":1,"privateElement":true,"label":"reportedBy","_class":"com.frontendart.codee.backend.entity.domain.edge.IssueReporter"},{"_key":"bdba3d091fff0a96b171c7b82d45d18bebc88e79d254973bd412ac9bf4f8d4a9","_id":"issueReporters/bdba3d091fff0a96b171c7b82d45d18bebc88e79d254973bd412ac9bf4f8d4a9","_from":"issues/f2b2b9bf97cbf375b9ab9d8e8c94e4011d11e55e38918f9cf1082d190f631e5e","_to":"contributors/cad412e1bd992f31cfdd612f74294110fdc5b97f391ca85c92ae2d54ba1a0658","_rev":"_hSSM9-2---","accountId":1,"privateElement":true,"label":"reportedBy","_class":"com.frontendart.codee.backend.entity.domain.edge.IssueReporter"},{"_key":"edd6389a4dfc2797bbcc6c13f9d4e847c925c9312d1e3458aea2205d9d825bc6","_id":"issueReporters/edd6389a4dfc2797bbcc6c13f9d4e847c925c9312d1e3458aea2205d9d825bc6","_from":"issues/a74b232ee49f5db1929457875b44f7a25b4eaa2246777a22e8c1561cc06c240e","_to":"contributors/8d064ec0aba42f0aa42d76b4ebd78b641a51ab76fa6b8676d6a95bd664f5af05","_rev":"_hSSM9XW---","accountId":1,"privateElement":true,"label":"reportedBy","_class":"com.frontendart.codee.backend.entity.domain.edge.IssueReporter"},{"_key":"2ca8969d7f721022ed949983524d8a7612933fe883f04a157c7b25160a7c811c","_id":"issueReporters/2ca8969d7f721022ed949983524d8a7612933fe883f04a157c7b25160a7c811c","_from":"issues/a74b232ee49f5db1929457875b44f7a25b4eaa2246777a22e8c1561cc06c240e","_to":"contributors/cad412e1bd992f31cfdd612f74294110fdc5b97f391ca85c92ae2d54ba1a0658","_rev":"_hSSM9ai---","accountId":1,"privateElement":true,"label":"reportedBy","_class":"com.frontendart.codee.backend.entity.domain.edge.IssueReporter"},{"_key":"2afe3cf18cba1abeaf0f9814e863eac05b1b841aed63e17ee2f6bc63148e6ffd","_id":"issueReporters/2afe3cf18cba1abeaf0f9814e863eac05b1b841aed63e17ee2f6bc63148e6ffd","_from":"issues/94f2d912342a6b1a541a968d2f73cd6aaddc0b9cdc066bba08fd643ac826d9ef","_to":"contributors/5f21a1996c0cf3c60b9edef2a94218220f687629536b985a0e8721d52dc4f80b","_rev":"_hSSN-ta---","accountId":5,"privateElement":true,"label":"reportedBy","_class":"com.frontendart.codee.backend.entity.domain.edge.IssueReporter"}],"hasMore":false,"cached":false,"extra":{"warnings":[],"stats":{"writesExecuted":0,"writesIgnored":0,"scannedFull":5,"scannedIndex":0,"filtered":0,"httpRequests":0,"executionTime":2.1056300101918168E-4,"peakMemoryUsage":0}},"error":false,"code":201}) - This is the correct response for the query I executed. It contains five instances of the edge, four of which are previously existing edges, and the edge newly added during the transaction. So it seems like the adding and the query do run transactionally, but still end up in a not found exception at the end. Later on in the logs there is a number of messages being sent and the last one before the exception in the logs is this: 2024-01-26 10:17:08,737 DEBUG [main] c.a.i.v.VstCommunicationSync: Send Message (id=1632, head=[1,1,"codee_backend",1,"/_api/document/issues/94f2d912342a6b1a541a968d2f73cd6aaddc0b9cdc066bba08fd643ac826d9ef",{},{"x-arango-driver":"JavaDriver/6.22.0 (JVM/17)","content-type":"application/x-velocypack","accept":"application/x-velocypack"}], body={}) - This references a document entity (an Issue), that is added during the transaction in the test shown above, and there is no x-arango-trx-id header. In the logs after this response, the exception I see is a 404 not found: 2024-01-26 10:17:08,739 DEBUG [main] com.arangodb.ArangoCollection: Response: 404, Error: 1202 - document not found - com.arangodb.ArangoDBException: Response: 404, Error: 1202 - document not found This to me means that, the adding and query works transactionally, but there are some additional internal queries being executed, that no longer run in transaction and therefore will lead to a not found exception, when querying an entity that was added in transaction. If you need any additional part of the logs, let me know.

[rashtao]

Thanks for clarifying, you are correct: the mapping layer has no awareness of the transaction.

This means that the queries issued to populate the relations properties (eg. fields annotated with @From and @To) are executed outside the transaction and will therefore not find the target document, since the transaction is not yet committed.

To work around it, you could either:

1. use the underlying Java driver (without Spring Data mapping)
2. do not use annotations for external references in Spring Data (like @From, @To, @Ref)

For example, here is a slightly modified version of your code snippet that is working:

    @Autowired
    private ArangoOperations arangoOperations;

    @Data
    @RequiredArgsConstructor
    @Document("entityAs")
    public static class EntityA {
        @Id
        @NonNull
        private final String id;

        @NonNull
        private final String name;

        @ArangoId
        private String arangoId;
    }

    @Data
    @RequiredArgsConstructor
    @Document("entityBs")
    public static class EntityB {
        @Id
        @NonNull
        private final String id;

        @NonNull
        private final String name;

        @ArangoId
        private String arangoId;
    }

    @Data
    @AllArgsConstructor
    @Edge("edgeEntities")
    public static class EdgeEntity {
        @Field("_from")
        public String entityA;

        @Field("_to")
        public String entityB;
    }

// ...

        StreamTransactionEntity transaction = db.beginStreamTransaction(new StreamTransactionOptions().writeCollections("entityAs", "entityBs", "edgeEntities"));
        EntityA entityA = new EntityA("id", "name");
        arangoOperations.insert(entityA, new DocumentCreateOptions().streamTransactionId(transaction.getId()));

        EntityB entityB = new EntityB("id", "name");
        arangoOperations.insert(entityB, new DocumentCreateOptions().streamTransactionId(transaction.getId()));

        EdgeEntity edgeEntity = new EdgeEntity(entityA.getArangoId(), entityB.getArangoId());
        arangoOperations.insert(edgeEntity, new DocumentCreateOptions().streamTransactionId(transaction.getId()));

        List<EntityA> a = arangoOperations.query("FOR entityA IN entityAs RETURN entityA",
                new AqlQueryOptions().streamTransactionId(transaction.getId()), EntityA.class).asListRemaining();

        assertThat(a)
                .hasSize(1)
                .contains(entityA);

        List<EntityB> b = arangoOperations.query("FOR entityB IN entityBs RETURN entityB",
                new AqlQueryOptions().streamTransactionId(transaction.getId()), EntityB.class).asListRemaining();

        assertThat(b)
                .hasSize(1)
                .contains(entityB);

        List<EdgeEntity> e = arangoOperations.query("FOR edgeEntity IN edgeEntities RETURN edgeEntity",
                new AqlQueryOptions().streamTransactionId(transaction.getId()), EdgeEntity.class).asListRemaining();

        assertThat(e)
                .hasSize(1)
                .contains(edgeEntity);

        db.commitStreamTransaction(transaction.getId());

[jurisitsm]

My team came to the conclusion that, though the solution you provided works, we would loose some functionality, because our business logic involves traversing edges using the @Relations annotation, when referencing edges from a node, as well as it's fetch type, and direction. In order to avoid the large task of refactoring and changing our implementation, we're looking for solutions to override or intercept methods, and requests made by the java driver, so that the transaction id would eventually end up being in the header of the queries, that are executed. However, because of the complexity of code, it is difficult the find the right solution for this, would there be any chance, that you could help identify, the most straightforward way of achieving this? It seems to me, that it is the EdgeFromResolver, and EdgeToResolver classes that use the implementation of the ArangoOperations' find method that does not accept a DocumentCreateOptions object, therefor preventing the transactionality of the query.

Your assistance would be greatly appreciated, Jurisits Marcell

[rashtao]

Exactly, the transaction is not propagated to the methods responsible to fetch the linked entities. You might want to have a look at the work in progress to support transactions here: https://github.com/arangodb/spring-data/pull/249

[jurisitsm]

Yes, this does looks like, it would be an easier way for implementing transactionality, than my attempts at simply passing the transaction id everywhere. Would it be possible for you to give an estimate of what time frame these changes could be used in? I see there are changes from 2022, to last week. I'm asking because this might have an effect on what direction we would like to go, with our implementation.

[rashtao]

Unfortunately due to internal prioritization, the related work has not been planned yet. Note that the PR linked above is a community contribution.

[jurisitsm]

Would it be possible for you to give an estimate of when this bug might be fixed? We don't seem to be able to work around this issue, and would prefer not to use an external community contribution.

[rashtao]

@jurisitsm FYI https://github.com/arangodb/spring-data/pull/303#issuecomment-2241044493