dyn filter can not handle the singleton input.

[st1page]

create materialized view if not exists "test_last_ingestion_time" as (
select 'table_a' as source, TO_TIMESTAMP('2024-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS') as last_ingestion_time
union all
select 'table_b' as source, TO_TIMESTAMP('2024-01-01 00:00:01', 'YYYY-MM-DD HH24:MI:SS') as last_ingestion_time
);

create materialized view if not exists "test_records" as (
select 1 as id, TO_TIMESTAMP('2024-01-01 00:00:01', 'YYYY-MM-DD HH24:MI:SS') as record_timestamp
union all
select 2 as id, TO_TIMESTAMP('2024-01-01 00:00:02', 'YYYY-MM-DD HH24:MI:SS') as record_timestamp
);

create materialized view if not exists "test_window" as (
with time_window as (
    select max(last_ingestion_time) as window_end
  from "test_last_ingestion_time"
)
select
  id
  from "test_records", time_window
  where record_timestamp >= window_end
);

thread 'rw-streaming' panicked at src/stream/src/from_proto/dynamic_filter.rs:42:18:
vnodes not set for dynamic filter
stack backtrace:
   0: rust_begin_unwind
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/std/src/panicking.rs:645:5
   1: core::panicking::panic_fmt
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/panicking.rs:72:14
   2: core::panicking::panic_display
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/panicking.rs:196:5
   3: core::panicking::panic_str
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/panicking.rs:171:5
   4: core::option::expect_failed
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/option.rs:1980:5
   5: core::option::Option<T>::expect
             at /rustc/e4c626dd9a17a23270bf8e7158e59cf2b9c04840/library/core/src/option.rs:894:21
   6: <risingwave_stream::from_proto::dynamic_filter::DynamicFilterExecutorBuilder as risingwave_stream::from_proto::ExecutorBuilder>::new_boxed_executor::{{closure}}
             at ./src/stream/src/from_proto/dynamic_filter.rs:40:13
   7: risingwave_stream::from_proto::create_executor::{{closure}}
             at ./src/stream/src/from_proto/mod.rs:131:5

[st1page]

change it to table, the bug will not happen

CREATE TABLE test_last_ingestion_time(
  source varchar,
  last_ingestion_time timestamp,
);

CREATE TABLE test_records(
  id int,
  record_timestamp timestamp,
);

insert into test_last_ingestion_time 
  select 'table_a' as source, TO_TIMESTAMP('2024-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS') as last_ingestion_time
  union all
  select 'table_b' as source, TO_TIMESTAMP('2024-01-01 00:00:01', 'YYYY-MM-DD HH24:MI:SS') as last_ingestion_time; 

insert into test_records
  select 1 as id, TO_TIMESTAMP('2024-01-01 00:00:01', 'YYYY-MM-DD HH24:MI:SS') as record_timestamp
  union all
  select 2 as id, TO_TIMESTAMP('2024-01-01 00:00:02', 'YYYY-MM-DD HH24:MI:SS') as record_timestamp

[st1page]

dev=> explain (distsql, verbose) create materialized view if not exists "test_last_ingestion_time" as (
select 'table_a' as source, TO_TIMESTAMP('2024-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS') as last_ingestion_time
union all
select 'table_b' as source, TO_TIMESTAMP('2024-01-01 00:00:01', 'YYYY-MM-DD HH24:MI:SS') as last_ingestion_time
);
NOTICE:  Your session timezone is UTC. It was used in the interpretation of timestamps and dates in your query. If this is unintended, change your timezone to match that of your data's with `set timezone = [timezone]` or rewrite your query with an explicit timezone conversion, e.g. with `AT TIME ZONE`.

                                                                                QUERY PLAN                                                                                 
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Fragment 0
 StreamMaterialize { columns: [source, last_ingestion_time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
 ├── tables: [ Materialize: 4294967294 ]
 ├── output: [ 'table_a':Varchar, $expr10001, _row_id ]
 ├── stream key: [ _row_id ]
 └── StreamValues { rows: [['table_a':Varchar, '2024-01-01 00:00:00+00:00':Timestamptz, 0:Int64], ['table_b':Varchar, '2024-01-01 00:00:01+00:00':Timestamptz, 1:Int64]] }
     ├── output: [ 'table_a':Varchar, $expr10001, _row_id ]
     └── stream key: [ _row_id ]
 
 Table 4294967294
 ├── columns: [ source: character varying, last_ingestion_time: timestamp with time zone, _row_id: bigint ]
 ├── primary key: [ $2 ASC ]
 ├── value indices: [ 0, 1, 2 ]
 ├── distribution key: []
 └── read pk prefix len hint: 1