Materialize: OOM issue reported by user

[umanwizard]

The following was reported by an external user:

this is one of the problem views that caused an OOM. The demo I picked was the https://flink.apache.org/2020/07/28/flink-sql-demo-building-e2e-streaming-application.html

root=> create materialized view top_categories as 
root-> select 
root-> count(1) as cnt,
root-> c.description as category_name
root-> from json_data_jtest as u left join category c
root-> on cast(data::jsonb->>'category_id' as int) = cast(c.category as int)
root-> where
root-> data::jsonb->>'behavior'='buy'
root-> group by c.description

It is kind of intuitive to do the join on a json key/value pair as in the query. In order to overcome this problem, I had to create a new view like below and use this in-lieu of the above. This adds another 4 steps to the process.

CREATE MATERIALIZED VIEW "materialize"."public"."category2" AS SELECT "category"::int AS "category_id", "description" AS "category_name" FROM "materialize"."public"."category"
CREATE MATERIALIZED VIEW "materialize"."public"."json_data_jtest" AS SELECT "data"::jsonb AS "data" FROM (SELECT "convert_from"("data", 'utf8') AS "data" FROM "materialize"."public"."jtest")


CREATE MATERIALIZED VIEW "materialize"."public"."json_data_jtest2" AS SELECT ("data"::jsonb ->> 'category_id')::int AS "category_id" FROM "materialize"."public"."json_data_jtest" WHERE "data"::jsonb ->> 'behavior' = 'buy'

CREATE MATERIALIZED VIEW "materialize"."public"."top_categories" AS SELECT "count"(1) AS "cnt", "c"."category_name" FROM "materialize"."public"."json_data_jtest2" AS "u" LEFT JOIN "materialize"."public"."category2" AS "c" ON "u"."category_id" = "c"."category_id" GROUP BY 2

[cuongdo]

@wangandi would this issue be fixed or improved by #4192?

[frankmcsherry]

@cuongdo I believe it would, yes.

[wangandi]

Yes it would be improved by #4192 in the sense the query involves a join on on cast(data::jsonb->>'category_id' as int) = cast(c.category as int), and #4192 will result in better plans when join keys are on expressions of columns. Though, #4192 may not help by that much because the input only has two inputs.

There may be other factors leading to the OOM that we could improve on. For example, there is a left join in the query, and likely, #4151 would help reduce memory from outer joins. It would be nice to get an EXPLAIN PLAN FOR... to see if there are other contributing factors to the OOM.

[frankmcsherry]

I'm pretty certain that our improved outer join planning does not activate in the case of expressions. It could probably be generalized to that. This query would probably also benefit from other analysis improvements, like perhaps the property that cast preserves primary keys, under some circumstances (perhaps not, depending on what it does on out of range values).

[awang]

@krishmanoh2 i think you spotted this one. now that #4192 has been merged can you see if the issue has been fixed?

[krishmanoh2]

I tested and it did an OOM again. This is how I am testing -

1. Create sources (json + kafka and csv)

2. Load data into kafka (2.5M messages). Stop the data generator.

3. CSV file is around 5K entries.

4. Create a MV on top of the source (not materializing the sources). I am testing to see if I can avoid materializing sources to reduce memory and also reduce number of steps needed to create the final view.

materialize=> explain
materialize-> select
materialize-> count(1) as cnt,
materialize-> c.description as category_name
materialize-> from
materialize-> (SELECT convert_from(data, 'utf8') AS data FROM materialize.public.src_user_behavior)
materialize-> u left join src_category c
materialize-> on cast(u.data::jsonb->>'category_id' as int) = cast(c.category as int)
materialize-> where
materialize-> data::jsonb->>'behavior'='buy'
materialize-> group by c.description
materialize-> ;

Optimized Plan

%0 = +
| Get materialize.public.src_user_behavior (u74) +
| Map convert_from(#0, "utf8") +
+
%1 = +
| Get %0 +
| Filter !(isnull(strtoi32((strtojsonb(#2) -> "category_id")))) +
| ArrangeBy (strtoi32((strtojsonb(#2) -> "category_id"))) +
+
%2 = +
| Get materialize.public.src_category (u73) +
| Filter !(isnull(strtoi32(#0))) +
+
%3 = +
| Join %1 %2 (= strtoi32(#3) strtoi32((strtojsonb(#2) -> "category_id"))) +
| | implementation = Differential %2 %1.(strtoi32((strtojsonb(#2) -> "category_id")))+
| | demand = (#2, #4) +
+
%4 = +
| Get %3 +
| Filter ("buy" = (strtojsonb(#2) -> "behavior")) +
| Project (#2..#5) +
+
%5 = +
| Get %3 +
| Distinct group=(#2) +
| Negate +
| Filter ("buy" = (strtojsonb(#0) -> "behavior")) +
+
%6 = +
| Get %0 +
| Filter ("buy" = (strtojsonb(#2) -> "behavior")) +
| Distinct group=(#2) +
+
%7 = +
| Union %5 %6 +
| Filter ("buy" = (strtojsonb(#0) -> "behavior")) +
| ArrangeBy (#0) +
+
%8 = +
| Get %0 +
| Filter ("buy" = (strtojsonb(#2) -> "behavior")) +
+
%9 = +
| Join %7 %8 (= #0 #3) +
| | implementation = Differential %8 %7.(#0) +
| | demand = (#0) +
| Map null, null, null +
| Project (#0, #4..#6) +
+
%10 = +
| Union %4 %9 +
| Filter ("buy" = (strtojsonb(#0) -> "behavior")) +
| Reduce group=(#2) +
| | agg count(1) +
| Project (#1, #0) +

(1 row)

[frankmcsherry]

I think this hasn't changed:

I'm pretty certain that our improved outer join planning does not activate in the case of expressions.

I suspect there is at least an outer join planning defect here, one that defaults back to the very primitive plan appropriate for arbitrary join conditions.

If it is easy to check, does it behave better with an inner join instead of a left join?

[krishmanoh2]

Yes, replacing left with inner is not causing an OOM.

Plan with inner join -

Optimized Plan

%0 = +
| Get materialize.public.src_user_behavior (u74) +
| Filter !(isnull(strtoi32((strtojsonb(convert_from(#0, "utf8")) -> "category_id")))), ("buy" = (strtojsonb(convert_from(#0, "utf8")) -> "behavior"))+
| Map convert_from(#0, "utf8") +
| ArrangeBy (strtoi32((strtojsonb(#2) -> "category_id"))) +
+
%1 = +
| Get materialize.public.src_category (u73) +
| Filter !(isnull(strtoi32(#0))) +
+
%2 = +
| Join %0 %1 (= strtoi32(#3) strtoi32((strtojsonb(#2) -> "category_id"))) +
| | implementation = Differential %1 %0.(strtoi32((strtojsonb(#2) -> "category_id"))) +
| | demand = (#4) +
| Reduce group=(#4) +
| | agg count(1) +
| Project (#1, #0) +

(1 row)

[wangandi]

What does the plan look like if you use the workaround in #4151?

If you use the workaround, the query should look like this:

CREATE VIEW subview AS
SELECT data FROM
(SELECT convert_from(data, 'utf8') AS data FROM materialize.public.src_user_behavior)
WHERE data::jsonb->>'behavior'='buy';

CREATE INDEX subview_idx ON subview(cast(u.data::jsonb->>'category_id' as int));

SELECT
count(1) as cnt,
c.description as category_name
from subview u left join src_category c
on cast(u.data::jsonb->>'category_id' as int) = cast(c.category as int)
group by c.description;

[krishmanoh2]

materialize=> CREATE VIEW subview AS
materialize-> SELECT data FROM
materialize-> (SELECT convert_from(data, 'utf8') AS data FROM materialize.public.src_user_behavior)
materialize-> WHERE data::jsonb->>'behavior'='buy';
CREATE VIEW
materialize=>
materialize=> CREATE INDEX subview_idx ON subview(cast(data::jsonb->>'category_id' as int));
CREATE INDEX
materialize=> explain SELECT
materialize-> count(1) as cnt,
materialize-> c.description as category_name
materialize-> from subview u left join src_category c
materialize-> on cast(u.data::jsonb->>'category_id' as int) = cast(c.category as int)
materialize-> group by c.description;

Optimized Plan

%0 = +
| Get materialize.public.subview (u97) +
| ArrangeBy (strtoi32((strtojsonb(#0) -> "category_id"))) +
+
%1 = +
| Get materialize.public.src_category (u73) +
| Filter !(isnull(strtoi32(#0))) +
+
%2 = +
| Join %0 %1 (= strtoi32(#1) strtoi32((strtojsonb(#0) -> "category_id"))) +
| | implementation = Differential %1 %0.(strtoi32((strtojsonb(#0) -> "category_id")))+
| | demand = (#0, #2) +
| Filter !(isnull(strtoi32((strtojsonb(#0) -> "category_id")))) +
+
%3 = +
| Get %2 +
+
%4 = +
| Get %2 +
| Distinct group=(#0) +
| Negate +
+
%5 = +
| Get materialize.public.subview (u97) +
| Distinct group=(#0) +
+
%6 = +
| Union %4 %5 +
| ArrangeBy (#0) +
+
%7 = +
| Get materialize.public.subview (u97) +
+
%8 = +
| Join %6 %7 (= #0 #1) +
| | implementation = Differential %7 %6.(#0) +
| | demand = () +
| Map null, null, null +
| Project (#0, #2..#4) +
+
%9 = +
| Union %3 %8 +
| Reduce group=(#2) +
| | agg count(1) +
| Project (#1, #0) +

(1 row)

[wangandi]

I'm pretty certain that our improved outer join planning does not activate in the case of expressions.

Do we have a ticket for this already? If yes, we should close this as a duplicate of that one. If not, I vote for renaming this issue to "activate improved outer join planning for expressions". The fact that the "improved outer join planning" can be further improved is already covered in #4151.

[frankmcsherry]

Well, #2732 is still open. Probably writing down more information there is a good start, but perhaps it should be closed as under-specific.

[awang]

is this issue safe to close?

[krishmanoh2]

Yes, I could not reproduce it anymore.