source split change may use stale cache

[xxchan]

          > For existing splits, we should use the latest offset from the cache. `target_splits` is from meta and contains the initial offset.

I'm unsure whether this is always correct when scaling. The ideas are similar to the documentation on the function cache_may_stale. The safest way should be always recovering states from the state store.

Originally posted by @BugenZhao in #14172 (comment)

[github-actions]

This issue has been open for 60 days with no activity. Could you please update the status? Feel free to continue discussion or close as not planned.

[xxchan]

When scaling out, the partition of the existing executors will likely shrink and becomes a subset of the previous one

This problem will not happen for source, since we ensure the memory states are always the same as assigned ones:

risingwave/src/stream/src/executor/source/source_executor.rs

Lines 301 to 322 in a3c6e48

	if split_changed {
	tracing::info!(
	actor_id = self.actor_ctx.id,
	state = ?target_state,
	"apply split change"
	);
	core.updated_splits_in_epoch
	.retain(|split_id, _| target_state.contains_key(split_id));
	let dropped_splits = core
	.latest_split_info
	.extract_if(|split_id, _| !target_state.contains_key(split_id))
	.map(|(_, split)| split)
	.collect_vec();
	if should_trim_state && !dropped_splits.is_empty() {
	// trim dropped splits' state
	core.split_state_store.trim_state(&dropped_splits).await?;
	}
	core.latest_split_info = target_state;