R2DBC TransientDataAccessResourceException: Row with Id [...] does not exist

As part of R2DBC learning process, i have a complex scenario of creating related entities as part of creating main entities. I am trying to create a Film entity and its related entities.

My request is something like this

{
"title" : "Star Wars: The Rise Of Skywalker",
"description": "In the riveting conclusion of the landmark Skywalker saga, new legends will be born-and the final battle for freedom is yet to come.",
"releaseYear": "2019",
"language": "English",
"rentalDuration": 7,
"rentalRate": "4.99",
"length": 165,
"replacementCost": "14.99",
"rating": "PG",
"specialFeaturesList": [ "SciFi", "Star Wars"],
"filmCategory": "Sci-Fi",
"actors":[
    {
        "firstName":"Daisy",
        "lastName": "Ridley"
    },
    {
        "firstName":"Carrie",
        "lastName":"Fisher"
    },
    {
        "firstName": "Oscar",
        "lastName": "Isaac"
    },
    {
         "firstName": "Adam",
        "lastName": "Driver"
    },
    {
          "firstName": "Mark",
        "lastName": "Hamill"
    },
    {
          "firstName": "John",
        "lastName": "Boyega"
    }
]

}

And below is my logic and code

   /**
 * Logic :
 * Transaction start
 * -- find Language or insert
 * -- find Flim by (Title,release year and languageId) or insert
 * -- foreach( actor)
 * ---- find or create actor
 * ---- find or create film_actor ( film id, actor)
 * -- end
 * -- for category
 * ---- find category or insert
 * ---- find film_category or insert
 * -- end
 * -- Map data
 * Transaction end
 *
 * @param filmRequest
 * @return Mono<Pair < Boolean, FilmModel>>
 */

Code :

@Override
public Mono<Pair<Boolean, FilmModel>> create(FilmRequest filmRequest) {

    return Mono.just(filmRequest)
            .flatMap(this::getOrCreateLanguage)
            .flatMap(languageInput -> getOrCreateFilm(filmRequest, languageInput.getLanguageId()))
            .flatMap(filmInput ->
                    Mono.zip(Mono.just(filmInput),
                            Flux.fromIterable(filmRequest.getActors())
                                    .flatMap(this::getOrCreateActor)
                                    .doOnNext(actorInput -> getOrCreateFilmActor(filmInput.getSecond().getFilmId(), actorInput.getActorId()))
                                    .collectList(),
                            Mono.just(filmRequest.getFilmCategory())
                                    .flatMap(this::getOrCreateCategory)
                                    .flatMap(category -> getOrCreateFilmCategory(filmInput.getSecond().getFilmId(), category.getCategoryId()))
                    ))
            .map(tuple -> {
                FilmModel filmModel = GenericMapper.INSTANCE.filmToFilmModel(tuple.getT1().getSecond());
                List<ActorModel> actorModelList = tuple.getT2().stream().map(GenericMapper.INSTANCE::actorToActorModel).collect(Collectors.toList());
                filmModel.setActors(actorModelList);
                return Pair.of(tuple.getT1().getFirst(), filmModel);
            }).as(operator::transactional);
}

And this is the failure I am getting:

TransientDataAccessResourceException: Failed to update table [film_category]. Row with Id [1006] does not exist.

https://gist.github.com/harryalto/d91e653ca1054038b766169142737f87

The Film entity ID is not persisted and because of which the dependent insert is failing.

I resolved the issue. The main issue was that for film_category and film_actor, the primary was composite of foriegn keys(i.e film_id from film, category_id from category/ actor_id from actor). I had to change the definition for the repositories for both something like this

@Repository
public interface FilmActorRepository extends 
ReactiveCrudRepository<FilmActor, FilmActorPK> {

    @Query("select * from film_actor  where film_id = :film_id")
    Flux<FilmActor> findAllByFilmId(@Param("film_id") final Long filmId);

    @Query("select * from film_actor  where film_id = :film_id and actor_id = :actor_id limit 1")
    Mono<FilmActor> findByFilmIdAndActorId(@Param("film_id") final Long filmId, @Param("actor_id") final Long actorId);

    @Modifying
    @Query("insert into film_actor (film_id, actor_id, last_update) values (:#{#filmActor.filmId}, :#{#filmActor.actorId}, now()) on conflict DO NOTHING")
    Mono<FilmActor> save(final FilmActor filmActor);
}

where FilmActorPK is defined as

@Value
@Builder
public class FilmActorPK implements Serializable {
    private long filmId;
    private long actorId;
}

I did similar exercise for the FilmCategoryRepository

@Repository
public interface FilmCategoryRepository extends 
ReactiveCrudRepository<FilmCategory, FilmCategoryPK> {

    @Query("select * from film_category  where category_id = :category_id and film_id = :film_id limit 1")
    Mono<FilmCategory> findFirstByCategoryIdAndFilmId(@Param("category_id") Long categoryId, @Param("film_id") Long filmId);

    @Query("select * from film_category   where   film_id = :film_id limit 1")
    Mono<FilmCategory> findFirstByFilmId(@Param("film_id") Long filmId);

    @Modifying
    @Query(
        "insert into film_category (film_id, category_id, last_update) 
       values (:#{#filmCategory.filmId}, :#{#filmCategory.categoryId}, 
       now()) on conflict DO NOTHING"
    )
    Mono<FilmCategory> save(final FilmCategory filmCategory);
}

where FilmCategoryPK is defined as

@Value
@Builder
public class FilmCategoryPK implements Serializable {
    private long filmId;
    private long categoryId;
}

The idea was taken from this issue here

This resolved my problem. I understand that R2DBC has different philosophy and there are things which can be done in R2DBC to bring it to parity to JPA like syntax and some which cannot be. But i am really enjoying the R2DBC as i do more exercises as part of my learning.