feat: Propagate the CreateMview mutation barrier to init executor epoch

[hzxa21]

What's changed and what's your intention?

After #399, barrier carriers the next epoch that a stateful executor will use for storage reads/writes. Therefore, executor needs to learn about the init epoch it uses for storage reads and writes before processing stream chunks. This PR addresses this issue by allowing the mutation barrier for mview creation to flow through the mview sub-graph before any data.

Changes include:

• Chain operator will propagate the initial epoch instead of swallow it
• Local barrier manager should not panic if the managed BarrierState is None
• Introduce ExecutorState to keep track of the stateful executor initialization and make sure the first barrier is a valid epoch.

Checklist

• I have written necessary docs and comments
• I have added necessary unit tests and integration tests

Refer to a related PR or issue link (optional)

[MrCroxx]

Generally LGTM.

[codecov]

Codecov Report

Merging #511 (5282d5d) into main (35d9ec8) will increase coverage by 0.03%.
The diff coverage is 91.32%.

@@             Coverage Diff              @@
##               main     #511      +/-   ##
============================================
+ Coverage     72.92%   72.95%   +0.03%     
  Complexity     2683     2683              
============================================
  Files           876      876              
  Lines         48900    48994      +94     
  Branches       1557     1557              
============================================
+ Hits          35658    35744      +86     
- Misses        12437    12445       +8     
  Partials        805      805              

Flag	Coverage Δ
java	62.54% <ø> (ø)
rust	77.35% <91.32%> (+0.03%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
rust/meta/src/hummock/hummock_manager.rs	68.59% <ø> (ø)
rust/meta/src/manager/epoch.rs	77.41% <ø> (ø)
rust/stream/src/executor/actor.rs	76.92% <ø> (ø)
rust/meta/src/barrier/mod.rs	57.14% <46.15%> (+0.11%)	⬆️
rust/stream/src/executor/test_utils.rs	81.42% <66.66%> (-0.93%)	⬇️
rust/stream/src/executor/barrier_align.rs	86.48% <75.00%> (+7.69%)	⬆️
rust/stream/src/executor/mod.rs	70.96% <77.41%> (+2.21%)	⬆️
rust/server/tests/table_v2_materialize.rs	95.23% <100.00%> (+0.09%)	⬆️
...st/stream/src/executor/aggregation/agg_executor.rs	85.61% <100.00%> (+0.21%)	⬆️
rust/stream/src/executor/chain.rs	57.40% <100.00%> (ø)
... and 17 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 35d9ec8...5282d5d. Read the comment docs.

[BugenZhao]

LocalSimpleAgg is stateless now in our system.

[hzxa21]

Looked at the codes. I would argue that LocalSimpleAgg is still stateful but the state is not persistent. I think it is okay to keep impl StatefulExecutor since it enables the initialization check.

[BugenZhao]

Suggested change

	pub trait StatefuleExecutor: Executor {
	pub trait StatefulExecutor: Executor {

[BugenZhao]

Suggested change

	/// Waiting for the first barrier
	INIT,
	/// Can read from and write to storage
	ACTIVE(u64),
	/// Waiting for the first barrier
	Init,
	/// Can read from and write to storage
	Active(u64),

[hzxa21]

Good catch. Will fix all issues mentioned above in a follow-up PR.