# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.

mode cockroach

## Test a plausibly correct recursive query.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE
    foo (a int, b int) AS (SELECT 1, 2 UNION SELECT a, 7 FROM bar),
    bar (a int) as (SELECT a FROM foo)
SELECT * FROM bar;
----
Explained Query:
  With Mutually Recursive
    cte l0 =
      Project (#1)
        Map (1)
          Distinct project=[#0] monotonic
            Union
              Project (#1)
                Map (7)
                  Get l0
              Constant
                - (2)
  Return
    Get l0

Target cluster: mz_catalog_server

EOF

## Test a nested recursive query.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE
    foo (a int8) AS (
        WITH MUTUALLY RECURSIVE
            bar (b int8) AS (
                SELECT * FROM (SELECT * FROM foo UNION ALL SELECT * FROM bar)
            )
        SELECT * FROM (SELECT * FROM foo EXCEPT ALL SELECT * FROM bar)
    )
SELECT * FROM foo;
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

## Test inlining at an inner nesting level. (database-issues#5581)
## `bar` is used only in `baz`, so it should be inlined. The inner WMR should have only one cte.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE
    foo (a int8) AS (
        WITH MUTUALLY RECURSIVE
            bar (b int8) AS (
                SELECT * FROM (SELECT * FROM foo UNION ALL SELECT * FROM baz)
            ),
            baz (b int8) AS (
                SELECT b + 3 FROM bar WHERE b > -5
            )
        SELECT * FROM (SELECT * FROM foo EXCEPT ALL SELECT * FROM baz)
    )
SELECT * FROM foo;
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

statement ok
CREATE TABLE t1(f0 int);

## Test inlining a cte from a `Let` into a `LetRec`.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH(raw, humanized expressions) AS VERBOSE TEXT FOR
WITH
  cte0 AS (
    SELECT * from t1 where f0 < 27
  )
SELECT * FROM
(
  WITH MUTUALLY RECURSIVE
    cnt (i int) AS (
      (SELECT f0 AS i FROM cte0)
      UNION
      SELECT i+1 FROM cnt)
  SELECT * FROM cnt
);
----
Explained Query:
  With Mutually Recursive
    cte l0 =
      Reduce group_by=[#0]
        Union
          Filter (#0{f0} < 27)
            ReadStorage materialize.public.t1
          Project (#1)
            Map ((#0{i} + 1))
              Get l0
  Return
    Get l0

Source materialize.public.t1
  filter=((#0{f0} < 27))

Target cluster: quickstart

EOF

## Test consolidation of not-really nested recursive query.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE
    foo (a int8) AS (
        WITH MUTUALLY RECURSIVE
            bar (b int8) AS (
                SELECT * FROM foo
            )
        SELECT * FROM (SELECT * FROM foo UNION ALL SELECT * FROM bar)
    )
SELECT * FROM foo;
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

## Test consolidation of independent recursive query blocks.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR SELECT * FROM (
    WITH MUTUALLY RECURSIVE
        foo (a int8) AS (SELECT DISTINCT a FROM foo)
    SELECT * FROM foo
)
UNION ALL
SELECT * FROM (
    WITH MUTUALLY RECURSIVE
        bar (a int8) AS (SELECT DISTINCT a FROM bar)
    SELECT * FROM bar
);
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

statement ok
CREATE TABLE edges (src int, dst int);

## Complex inlining and other things
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE
    label (node int, comp int) AS (
        SELECT dst, MIN(comp)
        FROM (
            SELECT dst, dst AS comp FROM edges
            UNION ALL
            SELECT edges.dst, label.comp
            FROM edges, label
            WHERE edges.src = label.node
        )
        GROUP BY dst
    )
SELECT round, COUNT(*) FROM (
    WITH MUTUALLY RECURSIVE
        relabel (node int, comp int, round int) AS (
            SELECT DISTINCT ON(node) node, comp, round
            FROM (
                SELECT node, comp, 0 as round FROM label
                UNION ALL
                SELECT edges.dst, relabel.comp, relabel.round + 1
                FROM edges, relabel
                WHERE edges.src = relabel.node
            )
            ORDER BY node, comp
        )
    SELECT round FROM relabel
)
GROUP BY round;
----
Explained Query:
  With Mutually Recursive
    cte l0 =
      Reduce group_by=[#0{dst}] aggregates=[min(#1{dst})]
        Union
          Project (#1{dst}, #1{dst})
            ReadStorage materialize.public.edges
          Project (#1{dst}, #3)
            Join on=(#0{src} = #2{node}) type=differential
              ArrangeBy keys=[[#0{src}]]
                Filter (#0{src}) IS NOT NULL
                  ReadStorage materialize.public.edges
              ArrangeBy keys=[[#0{node}]]
                Filter (#0{dst}) IS NOT NULL
                  Get l0
  Return
    With Mutually Recursive
      cte l1 =
        TopK group_by=[#0{dst}] order_by=[#1{min_dst} asc nulls_last] limit=1
          Union
            Map (0)
              Get l0
            Project (#1{dst}, #3, #5)
              Map ((#4{round} + 1))
                Join on=(#0{src} = #2{node}) type=differential
                  ArrangeBy keys=[[#0{src}]]
                    Filter (#0{src}) IS NOT NULL
                      ReadStorage materialize.public.edges
                  ArrangeBy keys=[[#0{node}]]
                    Filter (#0{dst}) IS NOT NULL
                      Get l1
    Return
      Reduce group_by=[#0] aggregates=[count(*)]
        Project (#2)
          Get l1

Source materialize.public.edges

Target cluster: quickstart

EOF

#### Tests for RECURSION LIMIT

## RECURSION LIMIT -- Simple test

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE (RECURSION LIMIT 10)
  cnt (i int) AS (
    SELECT 1 AS i
    UNION
    SELECT i+1 FROM cnt)
SELECT * FROM cnt;
----
Explained Query:
  With Mutually Recursive [recursion_limit=10]
    cte l0 =
      Distinct project=[#0] monotonic
        Union
          Project (#1)
            Map ((#0{i} + 1))
              Get l0
          Constant
            - (1)
  Return
    Get l0

Target cluster: mz_catalog_server

EOF

## RECURSION LIMIT -- RETURN AT RECURSION LIMIT

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE (RETURN AT RECURSION LIMIT 10)
  cnt (i int) AS (
    SELECT 1 AS i
    UNION
    SELECT i+1 FROM cnt)
SELECT * FROM cnt;
----
Explained Query:
  With Mutually Recursive [recursion_limit=10, return_at_limit]
    cte l0 =
      Distinct project=[#0] monotonic
        Union
          Project (#1)
            Map ((#0{i} + 1))
              Get l0
          Constant
            - (1)
  Return
    Get l0

Target cluster: mz_catalog_server

EOF

## RECURSION LIMIT -- ERROR AT RECURSION LIMIT

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE (ERROR AT RECURSION LIMIT 10)
  cnt (i int) AS (
    SELECT 1 AS i
    UNION
    SELECT i+1 FROM cnt)
SELECT * FROM cnt;
----
Explained Query:
  With Mutually Recursive [recursion_limit=10]
    cte l0 =
      Distinct project=[#0] monotonic
        Union
          Project (#1)
            Map ((#0{i} + 1))
              Get l0
          Constant
            - (1)
  Return
    Get l0

Target cluster: mz_catalog_server

EOF

## RECURSION LIMIT -- consolidation of independent recursive query blocks with different RECURSION LIMIT
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR SELECT * FROM (
    WITH MUTUALLY RECURSIVE (RECURSION LIMIT 5)
        foo (a int8) AS (SELECT DISTINCT a FROM foo)
    SELECT * FROM foo
)
UNION ALL
SELECT * FROM (
    WITH MUTUALLY RECURSIVE (RECURSION LIMIT 7)
        bar (a int8) AS (SELECT DISTINCT a - 2 FROM bar)
    SELECT * FROM bar
);
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

## RECURSION LIMIT -- consolidation of independent recursive query blocks with equal RECURSION LIMIT.
## We want to see RECURSION LIMIT printed at the block level rather than on each cte.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR SELECT * FROM (
    WITH MUTUALLY RECURSIVE (RECURSION LIMIT 27)
        foo (a int8) AS (SELECT DISTINCT a FROM foo)
    SELECT * FROM foo
)
UNION ALL
SELECT * FROM (
    WITH MUTUALLY RECURSIVE (RECURSION LIMIT 27)
        bar (a int8) AS (SELECT DISTINCT a - 2 FROM bar)
    SELECT * FROM bar
);
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

## ITERATION RECURSION -- consolidation of not-really nested recursive query.
## Here, the ITERATION RECURSION of the inner WMR is irrelevant, because it's not recursive.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE (RECURSION LIMIT 9)
    foo (a int8) AS (
        WITH MUTUALLY RECURSIVE (RECURSION LIMIT 3)
            bar (b int8) AS (
                SELECT * FROM foo
            )
        SELECT * FROM (SELECT * FROM foo UNION ALL SELECT * FROM bar)
    )
SELECT * FROM foo;
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

# ITERATION RECURSION -- a nested recursive query.
# With materialize#27389 this stopped testing a thing; see issue database-issues#8294.
query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE (RECURSION LIMIT 17)
    foo (a int8) AS (
        WITH MUTUALLY RECURSIVE (RECURSION LIMIT 11)
            bar (b int8) AS (
                SELECT * FROM (SELECT * FROM foo UNION ALL SELECT * FROM bar)
            )
        SELECT * FROM (SELECT * FROM foo EXCEPT ALL SELECT * FROM bar)
    )
SELECT * FROM foo;
----
Explained Query (fast path):
  Constant <empty>

Target cluster: mz_catalog_server

EOF

## ITERATION RECURSION -- Complex inlining and other things.

query T multiline
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE (RECURSION LIMIT 21)
    label (node int, comp int) AS (
        SELECT dst, MIN(comp)
        FROM (
            SELECT dst, dst AS comp FROM edges
            UNION ALL
            SELECT edges.dst, label.comp
            FROM edges, label
            WHERE edges.src = label.node
        )
        GROUP BY dst
    )
SELECT round, COUNT(*) FROM (
    WITH MUTUALLY RECURSIVE (RECURSION LIMIT 23)
        relabel (node int, comp int, round int) AS (
            SELECT DISTINCT ON(node) node, comp, round
            FROM (
                SELECT node, comp, 0 as round FROM label
                UNION ALL
                SELECT edges.dst, relabel.comp, relabel.round + 1
                FROM edges, relabel
                WHERE edges.src = relabel.node
            )
            ORDER BY node, comp
        )
    SELECT round FROM relabel
)
GROUP BY round;
----
Explained Query:
  With Mutually Recursive [recursion_limit=21]
    cte l0 =
      Reduce group_by=[#0{dst}] aggregates=[min(#1{dst})]
        Union
          Project (#1{dst}, #1{dst})
            ReadStorage materialize.public.edges
          Project (#1{dst}, #3)
            Join on=(#0{src} = #2{node}) type=differential
              ArrangeBy keys=[[#0{src}]]
                Filter (#0{src}) IS NOT NULL
                  ReadStorage materialize.public.edges
              ArrangeBy keys=[[#0{node}]]
                Filter (#0{dst}) IS NOT NULL
                  Get l0
  Return
    With Mutually Recursive [recursion_limit=23]
      cte l1 =
        TopK group_by=[#0{dst}] order_by=[#1{min_dst} asc nulls_last] limit=1
          Union
            Map (0)
              Get l0
            Project (#1{dst}, #3, #5)
              Map ((#4{round} + 1))
                Join on=(#0{src} = #2{node}) type=differential
                  ArrangeBy keys=[[#0{src}]]
                    Filter (#0{src}) IS NOT NULL
                      ReadStorage materialize.public.edges
                  ArrangeBy keys=[[#0{node}]]
                    Filter (#0{dst}) IS NOT NULL
                      Get l1
    Return
      Reduce group_by=[#0] aggregates=[count(*)]
        Project (#2)
          Get l1

Source materialize.public.edges

Target cluster: quickstart

EOF

## Make sure that we run `NormalizeLets` just before `CollectIndexRequests`.
## Regression test for https://github.com/MaterializeInc/database-issues/issues/6371

statement ok
CREATE TABLE t (a int2, b int4, c int8, d uint2, e uint4, f uint8, g text);

statement ok
SET cluster_replica = r1

statement ok
EXPLAIN OPTIMIZED PLAN WITH (humanized expressions) AS VERBOSE TEXT FOR
select
  subq_0."c2" as c0,
  ((select "attlen" from pg_catalog.pg_attribute limit 1 offset 4)
      ) # ((pg_catalog.abs(
        CAST((select "a" from public.t limit 1 offset 1)
           as int2))) # (case when ((cast(null as oid)) < (cast(null as oid)))
          or (((subq_0."c2" is not NULL)
              and ((numrange(0,0)) <= (numrange(0,0))))
            and ((((select "write_frontier" from mz_internal.mz_frontiers limit 1 offset 1)
                  ) <> ((select pg_catalog.max("time") from mz_introspection.mz_compute_import_frontiers_per_worker)
                  ))
              or (false))) then pg_catalog.abs(
          CAST((select "tgtype" from pg_catalog.pg_trigger limit 1 offset 6)
             as int2)) else pg_catalog.abs(
          CAST((select "tgtype" from pg_catalog.pg_trigger limit 1 offset 6)
             as int2)) end
        )) as c1,
  subq_0."c0" as c2
from
  (select
        ref_0."credits_per_hour" as c0,
        ref_0."credits_per_hour" as c1,
        ref_0."replica_id" as c2,
        ref_0."cluster_name" as c3,
        (select "id" from mz_introspection.mz_scheduling_elapsed_raw limit 1 offset 57)
           as c4
      from
        mz_internal.mz_cluster_replica_history as ref_0
      where ref_0."replica_id" is NULL
      limit 102) as subq_0
where (~ (select "replication_factor" from mz_catalog.mz_clusters limit 1 offset 5)
      ) <= (~ case when (case when (((((select "details" from mz_internal.mz_sink_statuses limit 1 offset 4)
                    ) >= ((select pg_catalog.jsonb_agg("id") from mz_introspection.mz_dataflow_operators)
                    ))
                or ((cast(0 as name)) >= (cast(null as name))))
              or (subq_0."c1" is NULL))
            and (((select pg_catalog.min("tgtype") from pg_catalog.pg_trigger)
                   is not NULL)
              and (true)) then (select "d" from public.t limit 1 offset 4)
             else (select "d" from public.t limit 1 offset 4)
             end
          ) <> (pg_catalog.mod(
          CAST(cast(coalesce((select pg_catalog.max("d") from public.t)
              ,
            (select "d" from public.t limit 1 offset 2)
              ) as uint2) as uint2),
          CAST((select "d" from public.t limit 1 offset 3)
             as uint2))) then (select "replication_factor" from mz_catalog.mz_clusters limit 1 offset 2)
         else (select "replication_factor" from mz_catalog.mz_clusters limit 1 offset 2)
         end
      )
limit 117;

## Ensure that we hoist WMR-invariant Let bindings, to avoid a `raw` modifier on arrangements
## that can be accessed through keys (and which do not otherwise require linear work).

statement ok
create table potato (a TEXT, b TEXT);

statement ok
create index on potato(a);

## The only thing that needs to stay true about what follows is that `potato` is used only
## as indexed access, and has `raw = false` to avoid decanting its contents.
query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE
walk(a TEXT, b TEXT) AS (
    SELECT a, b
    FROM potato
    WHERE a = 'russet'

    UNION

    SELECT potato.a, potato.b
    FROM potato
    INNER JOIN walk
    ON potato.a = walk.b
)
select * from walk;
----
Explained Query:
  With
    cte l0 =
      Get::PassArrangements materialize.public.potato
        raw=false
        arrangements[0]={ key=[#0{a}], permutation=id, thinning=(#1) }
        types=[text?, text?]
  Return
    With Mutually Recursive
      cte l1 =
        ArrangeBy
          input_key=[#0, #1]
          raw=true
          Reduce::Distinct
            val_plan
              project=()
            key_plan=id
            Union
              Join::Linear
                linear_stage[0]
                  lookup={ relation=0, key=[#0{a}] }
                  stream={ key=[#0], thinning=() }
                source={ relation=1, key=[#0] }
                Get::PassArrangements l0
                  raw=false
                  arrangements[0]={ key=[#0{a}], permutation=id, thinning=(#1) }
                  types=[text?, text?]
                ArrangeBy
                  raw=true
                  arrangements[0]={ key=[#0], permutation=id, thinning=() }
                  types=[text]
                  Constant
                    - ("russet")
              Join::Linear
                linear_stage[0]
                  lookup={ relation=1, key=[#0{b}] }
                  stream={ key=[#0{a}], thinning=(#1) }
                source={ relation=0, key=[#0{a}] }
                Get::PassArrangements l0
                  raw=false
                  arrangements[0]={ key=[#0{a}], permutation=id, thinning=(#1) }
                  types=[text?, text?]
                ArrangeBy
                  raw=true
                  arrangements[0]={ key=[#0{b}], permutation=id, thinning=() }
                  types=[text]
                  Get::Collection l1
                    project=(#1)
                    filter=((#1{b}) IS NOT NULL)
                    raw=true
    Return
      Get::PassArrangements l1
        raw=true

Used Indexes:
  - materialize.public.potato_a_idx (differential join, lookup)

Target cluster: quickstart

EOF