# 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.

#
# Test relaxation of the must_consolidate flag in LIR refinements
# for single-time dataflows (aka monotonic one-shot `SELECT`s).
# PR https://github.com/MaterializeInc/materialize/pull/19680
#

statement ok
CREATE TABLE t (a int, b int);

statement ok
INSERT INTO t VALUES (9, 1), (8, 2), (7, 3), (6, 4), (5, 5);

statement ok
DELETE FROM t WHERE b % 2 = 0;

# Get that is non-monotonic, must_consolidate
query II
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b FROM t
);
----
1
9

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        must_consolidate
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        Get::PassArrangements materialize.public.t
          raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Mfp on non-monotonic Get, must_consolidate
query II
SELECT MIN(a), MAX(b)
FROM (
    SELECT b, a
    FROM t
    WHERE a % 2 = 1
);
----
5
5

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(a), MAX(b)
FROM (
    SELECT b, a
    FROM t
    WHERE a % 2 = 1
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        must_consolidate
        val_plan=id
        key_plan
          project=()
        Get::Collection materialize.public.t
          raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t
  filter=((1 = (#0{a} % 2)))

Target cluster: quickstart

EOF

# FlatMap on non-monotonic Get, must_consolidate
query II
SELECT MIN(a), MAX(b)
FROM (
    SELECT b, generate_series(1, a) AS a
    FROM t
);
----
1
5

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(a), MAX(b)
FROM (
    SELECT b, generate_series(1, a) AS a
    FROM t
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        must_consolidate
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        FlatMap generate_series(1, #0{a}, 1)
          mfp_after
            project=(#1, #2)
          Get::PassArrangements materialize.public.t
            raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Union of non-monotonic Get's, must_consolidate
query II
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
    WHERE a % 2 = 1
    UNION ALL
    SELECT a, b
    FROM t
    WHERE a % 2 = 0
);
----
1
9

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
    WHERE a % 2 = 1
    UNION ALL
    SELECT a, b
    FROM t
    WHERE a % 2 = 0
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        must_consolidate
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        Union
          Get::Collection materialize.public.t
            project=(#0, #1)
            filter=((1 = #2))
            raw=true
          Get::Collection materialize.public.t
            project=(#0, #1)
            filter=((0 = #2))
            raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t
  project=(#0..=#2)
  map=((#0{a} % 2))

Target cluster: quickstart

EOF

# Threshold, no must_consolidate
query II
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
    WHERE a % 2 = 1
    EXCEPT ALL
    SELECT a, b
    FROM t
    WHERE a % 2 = 0
);
----
1
9

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
    WHERE a % 2 = 1
    EXCEPT ALL
    SELECT a, b
    FROM t
    WHERE a % 2 = 0
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        input_key=#0, #1
        Threshold::Basic ensure_arrangement={ key=[#0, #1], permutation=id, thinning=() }
          ArrangeBy
            raw=false
            arrangements[0]={ key=[#0, #1], permutation=id, thinning=() }
            types=[integer, integer?]
            Union consolidate_output=true
              Get::Collection materialize.public.t
                project=(#0, #1)
                filter=((1 = #2))
                raw=true
              Negate
                Get::Collection materialize.public.t
                  project=(#0, #1)
                  filter=((0 = #2))
                  raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t
  project=(#0..=#2)
  map=((#0{a} % 2))

Target cluster: quickstart

EOF

# Join on non-monotonic Get's, must_consolidate
statement ok
CREATE INDEX t_idx ON t(b);

query II
SELECT MIN(a), MAX(b)
FROM (
    SELECT *
    FROM (
        SELECT a, b
        FROM t
        WHERE a % 2 = 1
    ) JOIN (
        SELECT a, b
        FROM t
        WHERE a = 5
    ) USING (a, b)
);
----
5
5

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(a), MAX(b)
FROM (
    SELECT *
    FROM (
        SELECT a, b
        FROM t
        WHERE a % 2 = 1
    ) JOIN (
        SELECT a, b
        FROM t
        WHERE a = 5
    ) USING (a, b)
);
----
Explained Query:
  With
    cte l0 =
      Get::PassArrangements materialize.public.t
        raw=false
        arrangements[0]={ key=[#1{b}], permutation={#0: #1, #1: #0}, thinning=(#0) }
        types=[integer?, integer?]
    cte l1 =
      Reduce::Hierarchical
        aggr_funcs=[max]
        skips=[0]
        monotonic
        must_consolidate
        val_plan=id
        key_plan
          project=()
        Join::Linear
          linear_stage[0]
            closure
              project=(#0)
              filter=((#0{b}) IS NOT NULL AND (#1{a} = 5) AND (1 = (#1{a} % 2)) AND (#2{a} = 5))
            lookup={ relation=1, key=[#1{b}] }
            stream={ key=[#1{b}], thinning=(#0) }
          source={ relation=0, key=[#1{b}] }
          Get::PassArrangements l0
            raw=false
            arrangements[0]={ key=[#1{b}], permutation={#0: #1, #1: #0}, thinning=(#0) }
            types=[integer?, integer?]
          Get::PassArrangements l0
            raw=false
            arrangements[0]={ key=[#1{b}], permutation={#0: #1, #1: #0}, thinning=(#0) }
            types=[integer?, integer?]
  Return
    Union
      Get::Arrangement l1
        project=(#1, #0)
        map=(5)
        key=
        raw=false
        arrangements[0]={ key=[], permutation=id, thinning=(#0) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l1
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0) }
          Constant
            - ()

Used Indexes:
  - materialize.public.t_idx (differential join)

Target cluster: quickstart

EOF

statement ok
DROP INDEX t_idx;

# Join on `ArrangeBy`s that arrange in-query, unfortunately must_consolidate
# TODO(vmarcos): We set must_consolidate here because we do not know if the
# `raw` form will be used. If a join had a way to not use the `raw` collection,
# but only the arrangement built as part of the query (e.g., by attaching to its
# stream instead), then we would have the opportunity to turn must_consolidate
# off. The present analysis would need to be slightly extended then to detect if:
# (a) The input to `ArrangeBy` is not arranged; (b) The `raw` form is set to
# `false`; and (c) An arranged `form` is requested.
query II
SELECT MIN(a), MAX(b)
FROM (
    SELECT *
    FROM (
        SELECT a, b
        FROM t
        WHERE a % 2 = 1
    ) JOIN (
        SELECT a, b
        FROM t
        WHERE a = 5
    ) USING (a, b)
);
----
5
5

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(a), MAX(b)
FROM (
    SELECT *
    FROM (
        SELECT a, b
        FROM t
        WHERE a % 2 = 1
    ) JOIN (
        SELECT a, b
        FROM t
        WHERE a = 5
    ) USING (a, b)
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[max]
        skips=[0]
        monotonic
        must_consolidate
        val_plan=id
        key_plan
          project=()
        Join::Linear
          linear_stage[0]
            lookup={ relation=1, key=[#0{b}] }
            stream={ key=[#0{b}], thinning=() }
          source={ relation=0, key=[#0{b}] }
          ArrangeBy
            raw=true
            arrangements[0]={ key=[#0{b}], permutation=id, thinning=() }
            types=[integer]
            Get::Collection materialize.public.t
              project=(#1)
              filter=((1 = (#0{a} % 2)))
              raw=true
          ArrangeBy
            raw=true
            arrangements[0]={ key=[#0{b}], permutation=id, thinning=() }
            types=[integer]
            Get::Collection materialize.public.t
              project=(#1)
              raw=true
  Return
    Union
      Get::Arrangement l0
        project=(#1, #0)
        map=(5)
        key=
        raw=false
        arrangements[0]={ key=[], permutation=id, thinning=(#0) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0) }
          Constant
            - ()

Source materialize.public.t
  filter=((#0{a} = 5) AND (#1{b}) IS NOT NULL)

Target cluster: quickstart

EOF

# Reduce, with aggregates, is an enforcer, no must_consolidate
query II
SELECT MIN(b), MAX(sum_a)
FROM (
    SELECT b, SUM(a) AS sum_a
    FROM t
    GROUP BY b
);
----
1
9

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(b), MAX(sum_a)
FROM (
    SELECT b, SUM(a) AS sum_a
    FROM t
    GROUP BY b
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        val_plan=id
        key_plan
          project=()
        input_key=#0
        Reduce::Accumulable
          simple_aggrs[0]=(0, 0, sum(#0{a}))
          val_plan
            project=(#0)
          key_plan
            project=(#1)
          Get::PassArrangements materialize.public.t
            raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Reduce, no aggregates, is an enforcer, no must_consolidate
query II
SELECT MIN(a), MAX(b)
FROM (
    SELECT DISTINCT b, a FROM t
);
----
5
5

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(a), MAX(b)
FROM (
    SELECT DISTINCT b, a FROM t
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        input_key=#0, #1
        Reduce::Distinct
          val_plan
            project=()
          key_plan
            project=(#1, #0)
          Get::PassArrangements materialize.public.t
            raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Get, Union, FlatMap chained from Reduce enforcer, no must_consolidate
statement ok
CREATE VIEW single_time_monotonic_t AS
    SELECT b, a, COUNT(*) AS c FROM t GROUP BY b, a;

query III
SELECT * FROM single_time_monotonic_t;
----
1
9
1
3
7
1
5
5
1

query II
WITH input AS (
    SELECT a, b, c
    FROM single_time_monotonic_t
    WHERE a % 2 = 1
    UNION ALL
    SELECT a, b, c
    FROM single_time_monotonic_t
    WHERE c % 2 = 1
)
SELECT MIN(a), MAX(b)
FROM (
    SELECT b, generate_series(1, a) AS a
    FROM input
    WHERE a + 1 = 6 AND c + 1 = 2
);
----
1
5

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
WITH input AS (
    SELECT a, b, c
    FROM single_time_monotonic_t
    WHERE a % 2 = 1
    UNION ALL
    SELECT a, b, c
    FROM single_time_monotonic_t
    WHERE c % 2 = 1
)
SELECT MIN(a), MAX(b)
FROM (
    SELECT b, generate_series(1, a) AS a
    FROM input
    WHERE a + 1 = 6 AND c + 1 = 2
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Accumulable
        simple_aggrs[0]=(0, 0, count(*))
        val_plan
          project=(#2)
          map=(true)
        key_plan
          project=(#1, #0)
        mfp_after
          filter=((2 = (#2{c} + 1)))
        Get::Collection materialize.public.t
          raw=true
    cte l1 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        FlatMap generate_series(1, #1{a}, 1)
          mfp_after
            project=(#0, #2)
          Union
            Get::Arrangement l0
              project=(#0, #1)
              filter=((1 = (#1{a} % 2)))
              key=#0, #1
              raw=false
              arrangements[0]={ key=[#0, #1], permutation=id, thinning=(#2) }
            Get::Arrangement l0
              project=(#0, #1)
              filter=((1 = (#2{c} % 2)))
              key=#0, #1
              raw=false
              arrangements[0]={ key=[#0, #1], permutation=id, thinning=(#2) }
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l1
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l1
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t
  filter=((6 = (#0 + 1)))

Target cluster: quickstart

EOF

# Top-1 is an enforcer, no must_consolidate
query II
SELECT MIN(b), MAX(a)
FROM (
    SELECT DISTINCT ON(a) a, b
    FROM t
    ORDER BY a, b DESC
);
----
1
9

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(b), MAX(a)
FROM (
    SELECT DISTINCT ON(a) a, b
    FROM t
    ORDER BY a, b DESC
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        TopK::MonotonicTop1 group_by=[#0] order_by=[#1 desc nulls_first] must_consolidate
          Get::PassArrangements materialize.public.t
            raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Top-k is an enforcer, no must_consolidate
query II
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
    ORDER BY b DESC
    LIMIT 2
);
----
3
7

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MIN(b), MAX(a)
FROM (
    SELECT a, b
    FROM t
    ORDER BY b DESC
    LIMIT 2
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[min, max]
        skips=[0, 0]
        monotonic
        val_plan
          project=(#1, #0)
        key_plan
          project=()
        TopK::MonotonicTopK order_by=[#1 desc nulls_first] limit=2 must_consolidate
          Get::PassArrangements materialize.public.t
            raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
      Mfp
        project=(#0, #1)
        map=(null, null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0, #1) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Top-1 can have no must_consolidate, not only min/max
query II
SELECT DISTINCT ON(max_a) max_a, b
FROM (
    SELECT b, MAX(a) AS max_a
    FROM t
    GROUP BY b
)
ORDER BY max_a, b DESC;
----
5
5
7
3
9
1

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT DISTINCT ON(max_a) max_a, b
FROM (
    SELECT b, MAX(a) AS max_a
    FROM t
    GROUP BY b
)
ORDER BY max_a, b DESC;
----
Explained Query:
  Finish order_by=[#0 asc nulls_last, #1 desc nulls_first] output=[#0, #1]
    TopK::MonotonicTop1 group_by=[#0] order_by=[#1 desc nulls_first]
      Mfp
        project=(#1, #0)
        input_key=#0
        Reduce::Hierarchical
          aggr_funcs=[max]
          skips=[0]
          monotonic
          must_consolidate
          val_plan
            project=(#0)
          key_plan
            project=(#1)
          Get::PassArrangements materialize.public.t
            raw=true

Source materialize.public.t

Target cluster: quickstart

EOF

# Top-k can have no must_consolidate, not only min/max.
# In addition, the lack of need for must_consolidate
# propagates through Mfp to a higher-level min/max.
query I
SELECT MAX(max_a)
FROM (
    SELECT max_a, b
    FROM (
        SELECT b, MAX(a) AS max_a
        FROM t
        GROUP BY b
    )
    ORDER BY b DESC
    LIMIT 2
);
----
7

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MAX(max_a)
FROM (
    SELECT max_a, b
    FROM (
        SELECT b, MAX(a) AS max_a
        FROM t
        GROUP BY b
    )
    ORDER BY b DESC
    LIMIT 2
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[max]
        skips=[0]
        monotonic
        val_plan=id
        key_plan
          project=()
        Mfp
          project=(#1)
          TopK::MonotonicTopK order_by=[#0 desc nulls_first] limit=2
            ArrangeBy
              input_key=[#0]
              raw=true
              Reduce::Hierarchical
                aggr_funcs=[max]
                skips=[0]
                monotonic
                must_consolidate
                val_plan
                  project=(#0)
                key_plan
                  project=(#1)
                Get::PassArrangements materialize.public.t
                  raw=true
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0) }
      Mfp
        project=(#0)
        map=(null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0) }
          Constant
            - ()

Source materialize.public.t

Target cluster: quickstart

EOF

# Monotonic constant, no must_consolidate
query I
SELECT MAX(a)
FROM (
    SELECT generate_series(1, 20000) AS a
);
----
20000

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
SELECT MAX(a)
FROM (
    SELECT generate_series(1, 20000) AS a
);
----
Explained Query:
  With
    cte l0 =
      Reduce::Hierarchical
        aggr_funcs=[max]
        skips=[0]
        monotonic
        val_plan=id
        key_plan
          project=()
        FlatMap generate_series(1, 20000, 1)
          Constant
            - ()
  Return
    Union
      ArrangeBy
        input_key=[]
        raw=true
        Get::PassArrangements l0
          raw=false
          arrangements[0]={ key=[], permutation=id, thinning=(#0) }
      Mfp
        project=(#0)
        map=(null)
        Union consolidate_output=true
          Negate
            Get::Arrangement l0
              project=()
              key=
              raw=false
              arrangements[0]={ key=[], permutation=id, thinning=(#0) }
          Constant
            - ()

Target cluster: quickstart

EOF

# Recursive context: No refinement of monotonic operators in recursive terms,
# so must_consolidate does not even apply to those. For operators outside that
# consume from a recursive term, must_consolidate is set as we cannot guarantee
# monotonicity for now even if the term is morally monotonic.
query I
WITH MUTUALLY RECURSIVE input(a int) AS (
    SELECT generate_series(1, 40000) AS a
    UNION
    SELECT DISTINCT ON(a) a
    FROM input
    WHERE a > 20000
)
SELECT MAX(a)
FROM input;
----
40000

query T multiline
EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
WITH MUTUALLY RECURSIVE input(a int) AS (
    SELECT generate_series(1, 40000) AS a
    UNION
    SELECT DISTINCT ON(a) a
    FROM input
    WHERE a > 20000
)
SELECT MAX(a)
FROM input;
----
Explained Query:
  With Mutually Recursive
    cte l0 =
      ArrangeBy
        input_key=[#0]
        raw=true
        Reduce::Distinct
          val_plan
            project=()
          key_plan=id
          Union
            FlatMap generate_series(1, 40000, 1)
              Constant
                - ()
            TopK::Basic group_by=[#0] limit=1
              Get::Collection l0
                filter=((#0{a} > 20000))
                raw=true
  Return
    With
      cte l1 =
        Reduce::Hierarchical
          aggr_funcs=[max]
          skips=[0]
          monotonic
          must_consolidate
          val_plan=id
          key_plan
            project=()
          Get::PassArrangements l0
            raw=true
    Return
      Union
        ArrangeBy
          input_key=[]
          raw=true
          Get::PassArrangements l1
            raw=false
            arrangements[0]={ key=[], permutation=id, thinning=(#0) }
        Mfp
          project=(#0)
          map=(null)
          Union consolidate_output=true
            Negate
              Get::Arrangement l1
                project=()
                key=
                raw=false
                arrangements[0]={ key=[], permutation=id, thinning=(#0) }
            Constant
              - ()

Target cluster: quickstart

EOF