panic while using nested `jsonb_array_elements`

[kwannoel]

https://www.notion.so/risingwave-labs/using-nested-jsonb_array_elements-leads-to-panic-in-distribution-285f0d69cb7680a4a728fdc47985d2fb#285f0d69cb7680a4a728fdc47985d2fb

[BugenZhao]

I suspect the issue arises from using JSONB as the join key. The Eq or Hash implementations for JSONB can be unreliable, especially given that JSON's number is totally a mess.

---

Note that the final HashJoin is immediately following the previous Exchange, where the data was just shuffled based on the hash of the JSONB key, without any further processing. So it's very weird that later in HashJoin when we re-calculate the vnode again, we get a different value.

However, we always encode the key into HashKey and deserialize it back to Datum when performing a match:

risingwave/src/stream/src/executor/hash_join.rs

Line 903 in 7b42daa

let keys = K::build_many(&side_update.join_key_indices, chunk.data_chunk());

risingwave/src/stream/src/executor/join/hash_join.rs

Line 286 in 6f0cc16

let decoded_key = key.deserialize(join_key_data_types)?;

As a result, we actually recalculate the vnode based on a value that has undergone a roundtrip of value encoding of JSONB, which is, under the hood a pair of serde_json::to_string and serde_json::deserialize.

It then appears that certain values, such as very large floating-point numbers, cannot survive the roundtrip. So I'm suspecting if there's also such value in user's workload.

---

A minimal reproduction could be:

create table t (v varchar);
create materialized view jt as select v::jsonb as v from t;
create materialized view mv as select t1.v from jt as t1, jt as t2 where t1.v = t2.v;

insert into t values ('{"":555555155555572055555781056}');

We get:

thread 'rw-streaming' panicked at src/common/src/hash/table_distribution.rs:274:9:
vnode 95 should not be accessed by this table
vnode count: 256
allowed vnodes: 66..=87
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace


*** await tree context of current task ***

Actor 52: `mv` [2.004s]
  Epoch 9379770755973120 [1.001s]
    HashJoin 3400000003 [1.001s]  <== current

Note that we create a table with VARCHAR column instead of JSONB, to ensure that the value won't go through the value-encoding roundtrip during insertion, but only when it's about to be processed in the streaming pipeline.

cc @xiangjinwu