fix(streaming): fix index's stream key to include distribution key

[chenzl25]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

• Previously, index's stream key is the same as the primary key of the primary table, but the distribution keys of the index are the index columns. In streaming queries, it is important to include the distribution keys to the stream keys, so in this PR, we will add the distribution keys of indexes to the stream keys. And for the backward compatibility issue, we add a method stream_key() to the table catalog, so that we can always include the distribution keys to the stream keys. We should never use the stream_key field directly, instead we should use the stream_key() method in the future.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

[chenzl25]

The derivation of join keys logic is changed. We apply the i2o mapping at the end. Otherwise, index selection for streaming join might panic, because index selection will replace a primary table with an index. Before this PR, they will have the same stream keys, but after this PR, they have different stream keys. For other operator like Agg, TopN, we can safely replace it with indexes, because the index stream_key's existence can always be guaranteed after logical_rewrite_for_stream. However, for the join, we only keep one side of the join key as the join's stream_key, and the other side's join key could be removed (even if it is the stream_key of the index).

[chenzl25]

thread 'rw-streaming' (19144930) panicked at src/stream/src/executor/wrapper/update_check.rs:38:21:
assertion left == right failed: U- and U+ should have same stream key
U- row: [8852, 107730, 1, 5, 773, NULL]
U- key: [8852, 1, 5, 773, NULL]
U+ row: [6892, 107730, 1, 5, 773, NULL]
U+ key: [6892, 1, 5, 773, NULL]
stream key indices: [0, 2, 3, 4, 5]
executor: Materialize 13D00000003
left: Project { row: [Some(Decimal(Normalized(8852))), Some(Int32(107730)), Some(Int32(1)), Some(Int32(5)), Some(Int32(773)), None], indices: [0, 2, 3, 4, 5] }
right: Project { row: [Some(Decimal(Normalized(6892))), Some(Int32(107730)), Some(Int32(1)), Some(Int32(5)), Some(Int32(773)), None], indices: [0, 2, 3, 4, 5] }

The e2e test failed, because it is possible that ExchangeExecutor passes through the U-/U+ to a same actor if the vnode of the distribution key calculation is the same even though the distribution key is changed. But Exchange's distribution key should be part of the stream key of the downstream executor.

It means we need to change U-/U+ to D-/I+ in ExchangeExecutor when the columns of distribution key are changed instead of doing these changes when vnode are different.

[chenzl25]

thread 'rw-streaming' (19144930) panicked at src/stream/src/executor/wrapper/update_check.rs:38:21:
assertion left == right failed: U- and U+ should have same stream key
U- row: [8852, 107730, 1, 5, 773, NULL]
U- key: [8852, 1, 5, 773, NULL]
U+ row: [6892, 107730, 1, 5, 773, NULL]
U+ key: [6892, 1, 5, 773, NULL]
stream key indices: [0, 2, 3, 4, 5]
executor: Materialize 13D00000003
left: Project { row: [Some(Decimal(Normalized(8852))), Some(Int32(107730)), Some(Int32(1)), Some(Int32(5)), Some(Int32(773)), None], indices: [0, 2, 3, 4, 5] }
right: Project { row: [Some(Decimal(Normalized(6892))), Some(Int32(107730)), Some(Int32(1)), Some(Int32(5)), Some(Int32(773)), None], indices: [0, 2, 3, 4, 5] }

The e2e test failed, because it is possible that ExchangeExecutor passes through the U-/U+ to a same actor if the vnode of the distribution key calculation is the same even though the distribution key is changed. But Exchange's distribution key should be part of the stream key of the downstream executor.

It means we need to change U-/U+ to D-/I+ in ExchangeExecutor when the columns of distribution key are changed instead of doing these changes when vnode are different.

waiting for #23603

[BugenZhao]

• fix(streaming): fix index's stream key to include distribution key #23592 👈 (View in Graphite)
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[BugenZhao]

LGTM!