`LANGUAGE sql` UDF 的体中包含 `FROM`，在一个作业中被调用两次时，流式状态会膨胀 33 倍

CREATE FUNCTION g_from(j JSONB) RETURNS VARCHAR LANGUAGE sql AS $$ SELECT y ->> 'k1' FROM (SELECT j AS y) t $$; CREATE TABLE payloads (id INT PRIMARY KEY, payload JSONB); EXPLAIN CREATE MATERIALIZED VIEW mv_from AS SELECT id, g_from(payload) AS a, g_from(payload) AS b FROM payloads; StreamMaterialize { columns: [id, a, b, payloads.payload(hidden), payloads.payload#1(hidden), payloads.payload#2(hidden)], stream_key: [id, payloads.payload#2, payloads.payload], pk_columns: [id, payloads.payload#2, payloads.payload], pk_conflict: NoCheck } └─StreamExchange { dist: HashShard(payloads.id, payloads.payload, payloads.payload) } └─StreamHashJoin { type: LeftOuter, predicate: payloads.payload IS NOT DISTINCT FROM payloads.payload } ├─StreamExchange { dist: HashShard(payloads.payload) } └─StreamTableScan { table: payloads, columns: [id, payload] } └─StreamExchange { dist: HashShard(payloads.payload) } └─StreamFilter { predicate: (IsNotNull(payloads.payload) OR IsNotNull(payloads.payload)) } └─StreamHashJoin { type: FullOuter, predicate: payloads.payload IS NOT DISTINCT FROM payloads.payload } ├─StreamProject { exprs: [payloads.payload, JsonbAccessStr(payloads.payload, 'k1':Varchar) as $expr1] } └─StreamHashAgg { group_key: [payloads.payload], aggs: [count] } └─StreamExchange { dist: HashShard(payloads.payload) } └─StreamTableScan { table: payloads, columns: [payload, id] } └─StreamProject { exprs: [payloads.payload, JsonbAccessStr(payloads.payload, 'k1':Varchar) as $expr2] } └─StreamHashAgg { group_key: [payloads.payload], aggs: [count] } └─StreamExchange { dist: HashShard(payloads.payload) } └─StreamTableScan { table: payloads, columns: [payload, id] }