dyn filter can not handle the singleton input. about risingwave

[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