An empty result may be returned when using ReactiveNeo4jTemplate's findAll method with a Statement parameter. Kotlin + Coroutines

[yacosta738]

I have an issue with ReactiveNeo4jTemplate and the findAll method using a Statement. All the template methods work except this one, which returns an empty result even though the database has nodes that match the Statement. I have a test that updates all the elements that match certain criteria, but it fails because the findAll with Statement does not return any value, so it does not perform the update. However, the findAll without any search criteria finds all the elements.

This is the implementation of the method that updates any data in a Statement.

class Neo4jQueryableRepositoryAdapter<T : Any, ID : Any>(
    private val delegator: Neo4jRepository<T, ID>,
    clazz: KClass<T>
) : QueryableRepository<T, ID>, Repository<T, ID> by delegator {
    private val parser = Neo4jCriteriaParser(clazz)
 
// More code and functions here

override fun updateAll(
        criteria: Criteria,
        patch: Patch<T>,
        limit: Int?,
        offset: Long?,
        sort: Sort?
    ): Flow<T> {
        return parser.parse(criteria).let { delegator.updateAll(it, patch, limit, offset, sort) }
    }

// More code and functions here

}

The implementation of my Neo4jRepository interface is located in the SimpleNeo4jRepository class.

@Suppress("UNCHECKED_CAST")
class SimpleNeo4jRepository<T : Any, ID : Any>(
    private var template: ReactiveNeo4jTemplate,
    val clazz: KClass<T>,
    eventPublisher: EventPublisher? = null,
) : Neo4jRepository<T, ID> {

// More code and functions

override fun updateAll(
        criteria: Statement,
        patch: Patch<T>,
        limit: Int?,
        offset: Long?,
        sort: Sort?
    ): Flow<T> {
        return updateAll(criteria, patch.async(), limit, offset, sort)
    }

override fun updateAll(
        criteria: Statement,
        patch: SuspendPatch<T>,
        limit: Int?,
        offset: Long?,
        sort: Sort?
    ): Flow<T> {
        return findAll(criteria, limit, offset, sort)
            .map { update(it, patch) }
            .filterNotNull()
    }

override fun findAll(criteria: Statement?, limit: Int?, offset: Long?, sort: Sort?): Flow<T> {
            // Validate the limit and offset
            val l = limit ?: Int.MAX_VALUE
            val o = offset ?: 0

            // Perform custom search if criteria are specified
            val all = if (criteria != null) {
                println("QUERY ➡️ ${criteria.cypher}")
                template.findAll(criteria, clazz.java)
            } else {
                template.findAll(clazz.java)
            }.subscribeOn(Schedulers.parallel())
                .asFlow()

            // Apply limit and offset and return results
            return all.drop(o.toInt()).take(l)
    }

// More code and Functions

}

The configuration class is as follows

import com.astrum.data.annotation.ConverterScope
import org.neo4j.driver.AuthTokens
import org.neo4j.driver.Driver
import org.neo4j.driver.GraphDatabase
import org.springframework.context.ApplicationContext
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.core.convert.converter.GenericConverter
import org.springframework.core.env.Environment
import org.springframework.data.convert.ReadingConverter
import org.springframework.data.convert.WritingConverter
import org.springframework.data.neo4j.config.AbstractReactiveNeo4jConfig
import org.springframework.data.neo4j.core.convert.Neo4jConversions
import org.springframework.data.neo4j.repository.config.EnableReactiveNeo4jRepositories
import org.springframework.transaction.annotation.EnableTransactionManagement

@Configuration
@EnableReactiveNeo4jRepositories("com.astrum")
@EnableTransactionManagement
class Neo4jConfiguration(
    private val applicationContext: ApplicationContext,
    private val env: Environment
) : AbstractReactiveNeo4jConfig() {

    @Bean
    override fun neo4jConversions(): Neo4jConversions {
        val converters = applicationContext.getBeansOfType(GenericConverter::class.java)
            .values
            .filter {
                it.javaClass.annotations.any { annotation ->
                    annotation is WritingConverter || annotation is ReadingConverter
                }
            }
            .filter {
                val scope = it.javaClass.annotations.filterIsInstance<ConverterScope>()
                scope.isEmpty() || scope.any { converterScope -> converterScope.type == ConverterScope.Type.NEO4J }
            }
        return Neo4jConversions(converters)
    }
    /**
     * The driver to be used for interacting with Neo4j.
     *
     * @return the Neo4j Java driver instance to work with.
     */
    @Bean
    override fun driver(): Driver {
        val uri = env.getProperty("spring.data.neo4j.uri") ?: "bolt://localhost:7687"
        val username = env.getProperty("spring.data.neo4j.username") ?: "neo4j"
        val password = env.getProperty("spring.data.neo4j.password") ?: "password"
        return GraphDatabase.driver(uri, AuthTokens.basic(username, password))
    }
}

This is the test that is failing when I use findAll with Statement, if I change it to just use findAll without Statement the test passes because of course it finds the node (but that's not the idea)

@Test
    fun updateAllByName() = parameterized { personRepository ->
        val person = DummyPerson.create()
            .let { personRepository.create(it) }
        val patch = DummyPerson.create()
        
                // just for check
        var all = personRepository.findAll(where(Person::name).`is`(person.name)).toList()
        println("all persons ✅: $all")

        val updatedPersons = personRepository.updateAll(
            where(Person::name).`is`(person.name),
            Patch.with {
                it.name = patch.name
                it.age = patch.age
            }
        ).toList()
 
        // just for check
        all = personRepository.findAll(where(Person::name).`is`(person.name)).toList()
        println("all persons ❌: $all")

        assertEquals(1, updatedPersons.size)
        val updatedPerson = updatedPersons[0]
        assertEquals(person.id, updatedPerson.id)
        assertEquals(patch.name, updatedPerson.name)
        assertEquals(patch.age, updatedPerson.age)
        assertNotNull(updatedPerson.updatedAt)
    }

This code is to test if the findAll function was working properly with the Criteria to Statement Converter.

var all = personRepository.findAll(where(Person::name).`is`(person.name)).toList()
        println("all persons ✅: $all")

Excerpts from the prints

all persons ✅: [Person(name=arletha.dietric, age=67)]

QUERY ➡️ MATCH (n:Person) WHERE n.name = 'arletha.dietric' RETURN n

all persons ❌: [Person(name=arletha.dietric, age=67)]

[meistermeier]

Could you please provide more details on this? I am somehow lost to find the real problem.

1. Although you mention this is just the excerpt from the output, I still wonder if there are three QUERY outputs. (first findAll, updateAll with implicit findAll and second findAll)
2. What do you expect of the update? Considering that you are using DummyPerson.create() for person and patch, the data should be the same. The outcome of the ❌ is in this way not surprising to me.

[yacosta738]

The main problem is with searching for all elements based on a given criteria (for which I have a mechanism that converts regardless of the type of persistence used). The method for finding all elements without a criteria works well and retrieves all elements, but when I use a criteria, that method doesn't find any element.

override fun findAll(criteria: Statement?, limit: Int?, offset: Long?, sort: Sort?): Flow<T> {
        if (limit != null && limit <= 0) {
            return emptyFlow()
        }
        return criteria?.let {
            template.findAll(it, clazz.java)
                .subscribeOn(Schedulers.parallel())
                .asFlow()
        } ?: findAll()
    }

I use the findAll(criteria) method when I want to perform an update based on a certain criteria. In my tests, I wanted to demonstrate how it works with just findAll and not findAll(statement). That's why I made the comment, because I'm not entirely sure why it works with some methods and not others.

During testing, changing the call to use findAll without criteria and manually filtering it works. However, it is ideal for it to work with criteria.

[yacosta738]

The given query MATCH (n:Person) WHERE n.name = 'floyd.lednerF4E' RETURN n, collect(n) should return the node with the name 'floyd.lednerF4E' and collect it in a list. However, it is currently not returning any nodes even if the node with the given name exists.

[meistermeier]

But how do you come to the conclusion that there should be another or updated Person in the test database?
You are creating two identically (given the fact that I don't know what is happening in DummyPerson.create()) persons.
The first findAll returns the correct data for person.
The update, if it works, would patch person to patch and those are equal (as far as I can see).
The last findAll returns exactly what I would expect.
Could you provide a small reproducer that I can see you problem. I don't get it. Also I cannot see your test case using findAll without criteria. The first and last one having where(Person::name).is(person.name).

[yacosta738]

Hi @meistermeier the example tests is in this class

Perhaps in the example I have shown it is not well appreciated. When testing and debugging the application, I was able to verify that, although the criterion is converted into a correct Statement, the findAll (given a Statement) and find (given a Statement) continue to fail. I am not sure if I am instantiating the configuration correctly so that Neo4jTemplate works correctly with search criteria.

[meistermeier]

Now I at least understand how the DummyPersons can differ. (found the other/right class that is related to the initial code)
What makes me wonder is why the findAll does work with the criteria in the test, but not the updateAll.
I have near to zero experience with Kotlin's Flow behaviour but given the fact that all boils down to the reactor Flow type in the end, I think your problem is related to this stackoverflow answer https://stackoverflow.com/a/54217285 (for more details also the other answer is a good read).
My interpretation on this for your use-case is: You are deferring the findAll work to a Scheduler.parallel() and this works fine. Adding the update to the chain will create an immediate call to the update with 0 elements in the flow.
At least what I think might happen. I would try not to use the subscribeOn method here to ensure the flow works as expected, first.

[yacosta738]

Thank you very much. I will review the answer on StackOverflow and try not to use subscribeOn in that case. Thanks for your help, I already have a clue about where to go next.

[meistermeier]

As I said, this is just my interpretation and it might be also something else.
Would be great to get feedback if this was the root cause of the problem.

[yacosta738]

😌 After several attempts trying to avoid the use of Scheduler.parallel(), it still happens that, in the case of updateAll (which in turn calls findAll with certain criteria), the expected values are not found. I don't know if, in the intermediate process, which is called within a flow to cache the results, excessive use of parallel processing causes the call to end without finding the expected values (in this case, Persona). When debugging the application and entering the ReactiveNeo4jTemplate flow, when the criteria are applied, no values are found in the database. The strangest thing is that I have a similar implementation for SQL that works correctly.

Related Issue:

• Problem with ReactiveNeo4jTemplate
• StackOverFlow question

[yacosta738]

It seems that the issues are with the update and updateAll calls. In this case, I added two dummy calls to the findAll method in the unit tests and found the expected values.