source: source reader will stall after replacing if previous reader has finished

[BugenZhao]

The SelectWithStrategy maintains an internal state of termination for both sides. If one side is terminated, it'll never poll it in the future, for optimization.

match this.internal_state {
    InternalState::Start => {
        let next_side = (this.clos)(this.state);
        poll_inner(&mut this, next_side, cx)
    }
    InternalState::LeftFinished => match this.stream2.poll_next(cx) {
        Poll::Ready(None) => {
            *this.internal_state = InternalState::BothFinished;
            Poll::Ready(None)
        }
        a => a,
    },
    InternalState::RightFinished => match this.stream1.poll_next(cx) {
        Poll::Ready(None) => {
            *this.internal_state = InternalState::BothFinished;
            Poll::Ready(None)
        }
        a => a,
    },
    InternalState::BothFinished => Poll::Ready(None),
}

If the source reader has finished before split changing (or scaling), even if we replace it with a new active reader, the SelectWithStrategy still won't poll it anymore. So we can never get the chunks from this source! Actually, the futures has warned us about this.

Note that care must be taken to avoid tampering with the state of the stream which may otherwise confuse this combinator.

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

Lines 88 to 91 in 5d8465a

	/// Replace the source stream with a new one for given `stream`. Used for split change.
	pub fn replace_source_stream(&mut self, stream: BoxSourceWithStateStream) {
	*self.inner.get_mut().1 = Self::source_stream(stream).map_ok(Either::Right).boxed();
	}

Since we set the event num for the Nexmark source in the deterministic scaling test, and it will finish after emitting all events, this is also why we get incorrect results if we allow scaling source in the chaos test.

• test(scale): chaos test for basic nexmark #6062 (comment)

risingwave/src/tests/simulation_scale/src/ctl_ext.rs

Lines 101 to 103 in 5d8465a

	// FIXME: we should already support reschedule source fragment, but there might be a bug
	// of nexmark generator recovery.
	let has_source = identity_contains("StreamSource")(f);

[tabVersion]

If the source reader has finished before split changing (or scaling)

what does it mean "a source reader has finished"? The source always consumes an endless stream.

[BugenZhao]

If the source reader has finished before split changing (or scaling)

what does it mean "a source reader has finished"? The source always consumes an endless stream.

In production, yes. However, the nexmark source will "finish" if we set the event num for it.

#5657 (comment)