bug(batch): Batch scheduler panics if one side of the join has singleton exchange and none for the other side.

[chenzl25]

Describe the bug

dev=> create table t (id int);
create materialized view v as select count(*) cnt from t;
CREATE_TABLE
CREATE_MATERIALIZED_VIEW
dev=> explain select * from v join (select count(*) from t) T;
                      QUERY PLAN
------------------------------------------------------
 BatchNestedLoopJoin { type: Inner, predicate: true }
 ├─BatchScan { table: v, columns: [cnt] }
 └─BatchSimpleAgg { aggs: [count] }
   └─BatchExchange { order: [], dist: Single }
     └─BatchScan { table: t, columns: [] }
(5 rows)


// panic
dev=> select * from v join (select count(*) from t) T;

As you can see, mv v has Singleton distribution, so its required exchange/shuffle is removed. Currently our scheduler will panic on either local or distributed execution mode.

frontend.log (local execution mode)

...
thread 'risingwave-main' panicked at 'not implemented: not supported in local mode', src/frontend/src/scheduler/task_context.rs:62:9
stack backtrace:
Wed Jan  4 12:51:21 UTC 2023 [risedev]: Program exited with 139

frontend.log (distributed execution mode)

...
thread 'risingwave-main' panicked at 'no partition info for seq scan', src/frontend/src/scheduler/distributed/stage.rs:791:22

To Reproduce

create table t (id int);
create materialized view v as select count(*) cnt from t;
-- panic
select * from v join (select count(*) from t) T;

Expected behavior

No response

Additional context

No response

[kwannoel]

Looking into to it, but appreciate if anyone has ideas to share, haven't worked on scheduler before. (Feel free to re-assign to someone else, maybe I will take some time to fix as I am unfamiliar...).

[liurenjie1024]

1. For local execution mode, the physical plan is incorrect. We are supposed to insert an exchange before all table scan, and table scan should be pushed to compute node. The panic for local execution mode is caused by trying to execute table scan in frontend.
2. For distributed execution mode, we need to investigate why no partition found for table.

cc @kwannoel

[kwannoel]

For distributed execution mode, we need to investigate why no partition found for table.

In distributed execution mode table scan is also in root fragment, hence the error:

dev=> create table t (id int);
CREATE_TABLE
dev=> create materialized view v as select count(*) cnt from t;
CREATE_MATERIALIZED_VIEW
dev=> SET QUERY_MODE TO distributed;
SET_OPTION
dev=> explain select * from v join (select count(*) from t) T;
                      QUERY PLAN
------------------------------------------------------
 BatchNestedLoopJoin { type: Inner, predicate: true }
 ├─BatchScan { table: v, columns: [cnt] }
 └─BatchSimpleAgg { aggs: [sum0(count)] }
   └─BatchExchange { order: [], dist: Single }
     └─BatchSimpleAgg { aggs: [count] }
       └─BatchScan { table: t, columns: [] }
(6 rows)

I guess an exchange should be inserted either before or after NestedLoopJoin.

• I think the one way to fix it would be to fix require_additional_exchange_on_root, since currently it does not check if operators with multiple scan nodes all have exchange between them and root. It should do as its documentation says, and ensure singleton table scan is not in the root stage:
  

  risingwave/src/frontend/src/optimizer/mod.rs

  Lines 455 to 456 in c8323d5

  	/// As we always run the root stage locally, we should ensure that singleton table scan is not
  	/// the root stage. Returns `true` if we must insert an additional exchange to ensure this.

Will implement this in PR (#7240) first, but open to other ideas.