feat(streaming): source executor use table_id for keyspace prefix

[Li0k]

What's changed and what's your intention?

source executor use table_id for keyspace prefix

• use keyspace::table_root to replace keyspace::executor_root same as materialized_executor

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)

• https://github.com/singularity-data/risingwave/pull/2992/files

[hzxa21]

Good job! We can finally remove executor_root and shared_executor_root in Keyspace.

[codecov]

Codecov Report

Merging #3010 (7d76625) into main (043beb7) will not change coverage.
The diff coverage is 0.00%.

@@           Coverage Diff           @@
##             main    #3010   +/-   ##
=======================================
  Coverage   73.11%   73.11%           
=======================================
  Files         724      724           
  Lines       97174    97174           
=======================================
  Hits        71049    71049           
  Misses      26125    26125           

Flag	Coverage Δ
rust	73.11% <0.00%> (ø)

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

Impacted Files	Coverage Δ
src/stream/src/from_proto/source.rs	0.00% <0.00%> (ø)

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

[skyzh]

Need to hold this PR for a while. Previously, each source has its own keyspace. But now, we make them share one keyspace, which would cause problems.

The correct implementation should be like

let keyspace = Keyspace::table_root(store, &source_id).append(&executor_id.to_le_bytes());

[hzxa21]

Need to hold this PR for a while. Previously, each source has its own keyspace. But now, we make them share one keyspace, which would cause problems.

The correct implementation should be like

let keyspace = Keyspace::table_root(store, &source_id).append(&executor_id.to_le_bytes());

Why would it cause problems? I thought each source split of the same MV should use different user keys to store their state.

[skyzh]

Why would it cause problems? I thought each source split of the same MV should use different user keys to store their state.

Did we ensure that explicitly? Previously source executor is using executor_root, which is per-executor instead of per-operator (like hash join and hash agg).

[hzxa21]

Why would it cause problems? I thought each source split of the same MV should use different user keys to store their state.

Did we ensure that explicitly? Previously source executor is using executor_root, which is per-executor instead of per-operator (like hash join and hash agg).

@tabVersion can confirm.

I skim through the current implementation and it seems that the split implementations of the supported source already ensure that:

SplitImpl will put split id or partition id in the return value of SplitMetaData::id(), which is used as the source state key. (I originally thought PulsarSplit has a wrong implementation since it only put topic in SplitMetaData::id() but I realized that the topic name is actually a sub-topic name with partition id as the suffix if we use partitioned topic in pulsar.)

[skyzh]

That's cool. So we can use shared_executor_root for source already? Then I'd approve this PR.

[tabVersion]

Did we ensure that explicitly? Previously source executor is using executor_root, which is per-executor instead of per-operator (like hash join and hash agg).

thanks for fixing the misuse of executor_id. source split's assignment mechanism guarantees one split can only be assigned to one executor so the prefix is unique in parallel.