`LANGUAGE sql` UDF whose body contains `FROM` inflates streaming state ~33x when called twice in one job

Describe the bug

Describe the bug

Two materialized views over the same 200-row table produce identical output (200 rows each, zero differing rows), but one uses ~33x the state:

payload is a JSONB column averaging 1,542 bytes.

The cause: a LANGUAGE sql UDF whose body contains a FROM clause is inlined as a correlated subquery. When such a UDF is called two or more times in the same streaming job, the optimizer unnests it into StreamHashJoin, and the argument expression becomes the join key. It then lands in stream_key, so the whole JSONB value is materialized in the join's internal state — once per call site.

Calling it once produces a plain StreamProject. A UDF with the same semantics but no FROM in its body produces StreamProject no matter how many times it is called.

The plan

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] }

The table's primary key is id alone, but stream_key is [id, payloads.payload#2, payloads.payload] — the JSONB column appears in it twice. Two levels of StreamHashJoin plus two StreamHashAgg and three StreamExchange on the JSONB value appear for what is semantically a scalar projection.

Reproduction

-- Body contains FROM.
CREATE FUNCTION g_from(j JSONB) RETURNS VARCHAR
LANGUAGE sql AS $$ SELECT y ->> 'k1' FROM (SELECT j AS y) t $$;

-- Same result, no FROM.
CREATE FUNCTION g_plain(j JSONB) RETURNS VARCHAR
LANGUAGE sql AS $$ SELECT j ->> 'k1' $$;

-- A second function identical to g_from, to separate "same UDF" from "two call sites".
CREATE FUNCTION h_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);
-- 200 rows, payload averaging 1542 bytes

Counting StreamHashJoin in EXPLAIN CREATE MATERIALIZED VIEW ...:

So what matters is the number of call sites of a FROM-bodied UDF, not whether it is the same function.

What is wrong with it

1. The rewrite is invisible. The UDF looks like a scalar function at the SQL level. There is no reason to expect that adding a FROM to its body changes the physical plan of every job that calls it.

2. It only triggers at 2+ call sites, so it appears when an unrelated part of the query is edited.

3. There is a related restriction with the same root cause. A LANGUAGE sql UDF whose body contains FROM cannot be used as an argument to an aggregate:

   sql复制

   SELECT count(DISTINCT g_from(payload)) FROM payloads;
   -- Feature is not yet implemented: subquery inside aggregation calls
   
   SELECT count(DISTINCT g_plain(payload)) FROM payloads;
   -- 200

Expected behavior

Either inline such a UDF as a scalar expression when the subquery introduces no extra rows (as in every example above), or surface the rewrite — at CREATE FUNCTION or at EXPLAIN time — so that the cost is attributable. Silently multiplying state is very hard to trace back to a FROM in a function body.

Environment

• RisingWave v3.1.0-rc.1
• Distributed local cluster (meta / frontend / compute / 2 compactors)
• State measured via rw_catalog.rw_table_stats joined to rw_catalog.rw_internal_tables, after FLUSH and a 90-second settle

Error message/log

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