Description
As brought to light from someone on the Gitter channel, it appears there is a significant performance overhead for query flows that use the ReactiveCassandraTemplate. Notably this includes the reactive @'Repository classes.
The throughput test ran compared writing using ReactiveCassandraRepository::insert vs ReactiveCqlTemplate::execute (cql, args). Local testing gave the following results (writes per second):
- ReactiveCassandraRepository::insert 4000 writes/s
- ReactiveCqlTemplate::execute 7500 writes/s
As you can see there is a significant difference.
I have tried the test with prepared statements both enabled and disabled and it makes little difference. CPU profiling shows no hotspots for the repository inserts and no discernible difference in the overall profiles. Could it be the mapping layer that builds the statements? I will continue to dig into possibilities.
The test showing the throughput difference can be found here (credits to original author piddubnyi): https://github.com/samueldlightfoot/spring-data-cassandra-performnace
Here are the JProfiler snapshots for runs of both for anyone interested (I may be missing something in my analysis):
Spring Data Performance.zip
Entity:
@Table("snapshot")
@Value
@Builder
@EqualsAndHashCode(callSuper = false)
@RequiredArgsConstructor
public class SnapshotRecord {
@PrimaryKeyColumn(ordinal = 0, type = PrimaryKeyType.PARTITIONED)
long id;
@PrimaryKeyColumn(ordinal = 1, type = PrimaryKeyType.PARTITIONED)
short market;
@PrimaryKeyColumn(ordinal = 3, type = PrimaryKeyType.CLUSTERED)
Instant slot;
double value;
}
Repository:
public interface SnapshotRepository extends ReactiveCrudRepository<SnapshotRecord, Long> {
default Mono<Boolean> saveViaCql(ReactiveCqlOperations cqlOps, SnapshotRecord record) {
return cqlOps.execute(
"INSERT INTO snapshot (id, market,slot,value) VALUES (?,?,?,?) USING TIMESTAMP ?;",
ps -> {
return ps.bind(
record.getId(),
record.getMarket(),
record.getSlot(),
record.getValue(),
record.getSlot().toEpochMilli() * 1000
);
}
);
}
}
Runner:
Flux<SnapshotRecord> data = Flux.generate(Object::new, (state, sink) -> {
ThreadLocalRandom random = ThreadLocalRandom.current();
sink.next(
new SnapshotRecord(
random.nextLong(),
(short) random.nextInt(),
Clock.systemUTC().instant(),
random.nextDouble()
)
);
return state;
});
subscription = data
//.flatMap((SnapshotRecord record) -> repository.saveViaCql(cqlOps, record), 512, 2048)
.flatMap(repository::save, 512, 2048) //doing this runs almost 2x slower than previous line
.doOnNext(d -> success.incrementAndGet())
.onErrorContinue((throwable, object) -> fail.incrementAndGet())
.subscribe();