Panicked when processing: must be invertible

[lmatz]

Describe the bug

create table t1(c varchar, n varchar, id varchar, d varchar);
create table t2(c varchar, p varchar, id varchar, d varchar);

explain create materialized view hierarchy as
select array_agg(t1.n order by path_idx) from t1
join t2
on t1.c = 'abc'
and t2.c = 'abc'
cross join unnest((case when t2.p <> '' then (string_to_array(trim(t2.p, ','), ',') || t2.d) else ARRAY[t2.d] end)) WITH ORDINALITY AS path_cols(path_val, path_idx)
where path_val = t1.id;

I am doing above after with ordinality as is supported. #12273
The query transforms from a user query.

Please add it to the test after it is fixed, thank you

Error message/log

thread 'risingwave-main' panicked at src/frontend/src/optimizer/plan_node/logical_scan.rs:240:10:
must be invertible
stack backtrace:
   0: rust_begin_unwind
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/std/src/panicking.rs:619:5
   1: core::panicking::panic_fmt
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/panicking.rs:72:14
   2: core::panicking::panic_display
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/panicking.rs:168:5
   3: core::panicking::panic_str
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/panicking.rs:152:5
   4: core::option::expect_failed
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/option.rs:1988:5
   5: core::option::Option<T>::expect
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/option.rs:898:21
   6: risingwave_frontend::optimizer::plan_node::logical_scan::LogicalScan::predicate_pull_up
             at ./src/frontend/src/optimizer/plan_node/logical_scan.rs:235:27
   7: <risingwave_frontend::optimizer::plan_node::logical_scan::LogicalScan as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_scan.rs:526:51
   8: <risingwave_frontend::optimizer::plan_node::logical_agg::LogicalAgg as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_agg.rs:1120:28
   9: risingwave_frontend::optimizer::plan_node::convert::ToStream::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/convert.rs:57:19
  10: <risingwave_frontend::optimizer::plan_node::logical_project::LogicalProject as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/logical_project.rs:246:25
  11: <risingwave_frontend::optimizer::plan_node::logical_project::LogicalProject as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_project.rs:256:9
  12: <risingwave_frontend::optimizer::plan_node::logical_project_set::LogicalProjectSet as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_project_set.rs:394:25
  13: risingwave_frontend::optimizer::plan_node::convert::ToStream::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/convert.rs:57:19
  14: risingwave_frontend::optimizer::plan_node::logical_join::LogicalJoin::to_stream_hash_join
             at ./src/frontend/src/optimizer/plan_node/logical_join.rs:871:25
  15: <risingwave_frontend::optimizer::plan_node::logical_join::LogicalJoin as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_join.rs:1357:17
  16: risingwave_frontend::optimizer::plan_node::convert::ToStream::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/convert.rs:57:19
  17: risingwave_frontend::optimizer::plan_node::logical_join::LogicalJoin::to_stream_hash_join
             at ./src/frontend/src/optimizer/plan_node/logical_join.rs:885:24
  18: <risingwave_frontend::optimizer::plan_node::logical_join::LogicalJoin as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_join.rs:1357:17
  19: risingwave_frontend::optimizer::plan_node::convert::ToStream::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/convert.rs:57:19
  20: <risingwave_frontend::optimizer::plan_node::logical_project::LogicalProject as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/logical_project.rs:246:25
  21: <risingwave_frontend::optimizer::plan_node::logical_project::LogicalProject as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_project.rs:256:9
  22: <risingwave_frontend::optimizer::plan_node::logical_agg::LogicalAgg as risingwave_frontend::optimizer::plan_node::convert::ToStream>::to_stream
             at ./src/frontend/src/optimizer/plan_node/logical_agg.rs:1120:28
  23: risingwave_frontend::optimizer::plan_node::convert::ToStream::to_stream_with_dist_required
             at ./src/frontend/src/optimizer/plan_node/convert.rs:57:19
  24: risingwave_frontend::optimizer::PlanRoot::gen_stream_plan
             at ./src/frontend/src/optimizer/mod.rs:395:28
  25: risingwave_frontend::optimizer::PlanRoot::gen_optimized_stream_plan
             at ./src/frontend/src/optimizer/mod.rs:298:24
  26: risingwave_frontend::optimizer::PlanRoot::gen_materialize_plan
             at ./src/frontend/src/optimizer/mod.rs:513:27
  27: risingwave_frontend::handler::create_mv::gen_create_mv_plan
             at ./src/frontend/src/handler/create_mv.rs:116:9
  28: risingwave_frontend::handler::explain::do_handle_explain::{{closure}}
             at ./src/frontend/src/handler/explain.rs:121:26
  29: risingwave_frontend::handler::explain::handle_explain::{{closure}}
             at ./src/frontend/src/handler/explain.rs:250:64
  30: risingwave_frontend::handler::handle::{{closure}}
             at ./src/frontend/src/handler/mod.rs:185:82
  31: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/future/future.rs:125:9
  32: <&mut F as core::future::future::Future>::poll
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/future/future.rs:113:9
  33: <tokio::time::timeout::Timeout<T> as core::future::future::Future>::poll
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.32.0/src/time/timeout.rs:202:33
  34: <risingwave_frontend::session::SessionImpl as pgwire::pg_server::Session>::run_one_query::{{closure}}
             at ./src/frontend/src/session.rs:945:88
  35: pgwire::pg_protocol::PgProtocol<S,SM>::inner_process_query_msg_one_stmt::{{closure}}
             at ./src/utils/pgwire/src/pg_protocol.rs:495:14
  36: <tracing::instrument::Instrumented<T> as core::future::future::Future>::poll
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tracing-0.1.37/src/instrument.rs:272:9
  37: pgwire::pg_protocol::PgProtocol<S,SM>::inner_process_query_msg::{{closure}}
             at ./src/utils/pgwire/src/pg_protocol.rs:477:18
  38: pgwire::pg_protocol::PgProtocol<S,SM>::process_query_msg::{{closure}}
             at ./src/utils/pgwire/src/pg_protocol.rs:435:65
  39: pgwire::pg_protocol::PgProtocol<S,SM>::do_process_inner::{{closure}}
             at ./src/utils/pgwire/src/pg_protocol.rs:271:88
  40: <core::panic::unwind_safe::AssertUnwindSafe<F> as core::future::future::Future>::poll
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/panic/unwind_safe.rs:296:9
  41: <futures_util::future::future::catch_unwind::CatchUnwind<Fut> as core::future::future::Future>::poll::{{closure}}
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.28/src/future/future/catch_unwind.rs:36:42
  42: <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/core/src/panic/unwind_safe.rs:271:9
  43: std::panicking::try::do_call
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/std/src/panicking.rs:526:40
  44: ___rust_try
  45: std::panicking::try
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/std/src/panicking.rs:490:19
  46: std::panic::catch_unwind
             at /rustc/62ebe3a2b177d50ec664798d731b8a8d1a9120d1/library/std/src/panic.rs:142:14
  47: <futures_util::future::future::catch_unwind::CatchUnwind<Fut> as core::future::future::Future>::poll
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.28/src/future/future/catch_unwind.rs:36:9
  48: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.32.0/src/task/task_local.rs:347:31
  49: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.32.0/src/task/task_local.rs:217:19
  50: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/martin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.32.0/src/task/task_local.rs:343:19
  51: pgwire::pg_protocol::PgProtocol<S,SM>::do_process::{{closure}}
             at ./src/utils/pgwire/src/pg_protocol.rs:192:14
  52: pgwire::pg_protocol::PgProtocol<S,SM>::process::{{closure}}
             at ./src/utils/pgwire/src/pg_protocol.rs:183:30

### To Reproduce

_No response_

### Expected behavior

_No response_

### How did you deploy RisingWave?

_No response_

### The version of RisingWave

_No response_

### Additional context

_No response_

[lmatz]

A side question

t1.c = 'abc'
and t2.c = 'abc'

is a nested loop join in PG, not RW:

risingwave/src/frontend/planner_test/tests/testdata/output/lateral_subquery.yaml

Line 192 in ca5e39e

└─StreamHashJoin { type: Inner, predicate: t2.p IS NOT DISTINCT FROM t2.p AND t2.p IS NOT DISTINCT FROM t2.p AND t2.d IS NOT DISTINCT FROM t2.d AND t2.d IS NOT DISTINCT FROM t2.d, output: [t1.n, t1.id, projected_row_id, t2.p, t2.p, t2.d, t2.d, Unnest(Case(($1 <> '':Varchar), ArrayAppend(StringToArray(Trim($1, ',':Varchar), ',':Varchar), $3), Array($3))), t2.p, t2.d, t1._row_id, t2._row_id] }

Is it difficult to make the optimizer deduce to be hash join by itself?

We can manually add t1.c = t2.c though, but considering we don't support nested loop in streaming queries, shall RW do it automatically?

Or produce some hints, no matter query-specific or general guidance, when nested loop join occurs in stream queries...