Tracking: setup micro benchmark for stream executors with in-memory state

[lmatz]

As mentioned in #5227, the purpose is to measure the pure computing performance, and a prerequisite is an efficient in-memory store.

As mentioned in #5227 (comment) by @BugenZhao , we now don't need to wait for a new in-memory store, unless there shows a performance bottleneck on the current in-memory store implementation.

• stream: Setup benchmark for HashAggExecutor #5683
• stream: Setup benchmark for HashJoinExecutor
• stream: Setup benchmark for LocalSimpleAggExecutor
• stream: Setup benchmark for GlobalSimpleAggExecutor
• stream: Setup benchmark for DynamicFilterExecutor
• stream: Setup benchmark for DispatchExecutor
• stream: Setup benchmark for FilterExecutor
• stream: Setup benchmark for ProjectSetExecutor
• stream: Setup benchmark for ExpandExecutor
• stream: Setup benchmark for BatchQueryExecutor
• stream: Setup benchmark for ChainExecutor
• stream: Setup benchmark for RearrangedChainExecutor
• stream: Setup benchmark for HopWindowExecutor
• stream: Setup benchmark for LookupExecutor
• stream: Setup benchmark for LookupUnionExecutor
• stream: Setup benchmark for UnionExecutor
• stream: Setup benchmark for SourceExecutor
• stream: Setup benchmark for MaterializeExecutor

Depends on #6285

[jon-chuang]

To my understanding, the current impl is a per-task store:

risingwave/src/storage/src/memory.rs

Line 171 in b0fb507

inner: Arc<RwLock<BTreeMap<KeyWithEpoch, Option<Bytes>>>>,

which is probably as good as one can do. But we won't be able to scale up/down or test recovery with this impl, but its not our objective.

Or, are we using shared version of MemoryStateStore? This would be bad as it would then be single lock.

So, to confirm, we are using per-task store for purposes of bench, correct?

[BugenZhao]

The shared one is a singleton, which can be used to simulate shared storage when running multiple compute nodes in a single process with risedev p.

For other cases, we're using the one constructed here whose lifetime is the same as compute_node_serve and is shared by all executors in this compute node.

risingwave/src/compute/src/server.rs

Line 111 in 833358c

let state_store = StateStoreImpl::new(

[jon-chuang]

😥 I see, it would be nice if we can create one that is spawned on a per-thread or per task basis, so we don't have to worry about contention at all... To my understanding, that is the objective of these in-memory benchmarks? I.e. to test purely the compute performance up to data serialization/deserialization?

[BugenZhao]

If we bench a single operator with the style of integrated tests in a single thread, there'll also be no contention. IIRC, the storage team is working on a refactoring of the local state store, we may check if we can improve it then.

[fuyufjh]

Removed from the milestone. Do it later.

[kwannoel]

Will this be priority again? Since we are looking at performance of stream engine?
Performance dashboard runs daily, whereas these benchmarks can be easily run ad-hoc to see if certain optimizations work or not.
Additionally we can generate flamegraph and see mem and cpu cost centres.