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

# https://github.com/MaterializeInc/advent-of-code-2023/blob/main/week1/aoc_1220.md

mode cockroach

statement ok
CREATE TABLE input (input TEXT);

# no input data

query T multiline
EXPLAIN OPTIMIZED PLAN WITH(humanized expressions, arity, join implementations) AS VERBOSE TEXT FOR
WITH MUTUALLY RECURSIVE

    lines(line TEXT) AS ( SELECT regexp_split_to_table(input, '\n') FROM input ),
    links(name TEXT, link TEXT) AS (
        SELECT
            substring(regexp_split_to_array(line, ' ')[1], 2),
            trim(',' FROM regexp_split_to_array(line, ' ')[x])
        FROM
            lines, generate_series(3, array_length(regexp_split_to_array(line, ' '), 1)) x
    ),
    -- One special line has op 'b' and name 'roadcaster'.
    types(op TEXT, name TEXT) AS (
        SELECT
            substring(regexp_split_to_array(line, ' ')[1], 1, 1),
            substring(regexp_split_to_array(line, ' ')[1], 2)
        FROM
            lines
    ),

    -- Part one: simulate 1000 steps of 'broadcaster' being activated with a low pulse.
    -- tally up total low and high pulses, and then multiply.
    -- The state carried across steps are the last-transmitted pulses of each operator.
    -- This should also tell us the final state of the `%` operators.
    -- We'll also need the totals of low and high pulses, so that we can add them up.

    seed(press INT, counter INT) AS (
        SELECT 1, 1
        UNION
        SELECT press, counter - 1
        FROM seed
        WHERE counter > 0
        UNION
        SELECT press + 1, 20
        FROM seed
        WHERE counter = 0
          AND press < 4100
    ),

    -- Emitted pulses after various button presses, in various rounds of resolution.
    pulses(name TEXT, press INT, round INT, pulse TEXT) AS (
        -- One thousand button presses, each followed by rounds of resolution.
        SELECT 'roadcaster', press, 1, 'lo' FROM seed WHERE counter = 0
        UNION ALL SELECT * FROM flip
        UNION ALL SELECT * FROM conj
    ),

    -- Counters; every 'lo' input pulse flips and emits the state.
    flip(name TEXT, press INT, round INT, pulse TEXT) AS (
        -- Each `signal` needs to behave as if all "prior" signals have been processed, ordered by (press, round, source).
        SELECT
            name,
            press,
            round + 1,
            -- Look for the most recently emitted signal, and we'll produce the opposite of that one.
            CASE WHEN (
                    SELECT COUNT(*)
                    FROM signal s1
                    WHERE s1.target = types.name
                      AND s1.pulse = 'lo'
                      AND ((s1.press < signal.press) OR
                           (s1.press = signal.press AND s1.round < signal.round) OR
                           (s1.press = signal.press AND s1.round = signal.round AND s1.source < signal.source))
                ) % 2 = 0
                THEN 'hi'
                ELSE 'lo'
            END
        FROM signal, types
        WHERE signal.target = types.name
            AND types.op = '%'
            AND signal.pulse = 'lo'
    ),

    -- NAND gates; every input pulse evokes the NAND of most recent inputs.
    conj(name TEXT, press INT, round INT, pulse TEXT) AS (
        SELECT
            name,
            press,
            round + 1,
            -- Look for the most recently received signals from each input,
            -- including this one, and iff all 'hi' then 'lo'.
            CASE WHEN (
                    (SELECT COUNT(*) FROM links WHERE link = types.name)
                    =
                    (SELECT COUNT(*) FROM (
                        SELECT DISTINCT ON (source) source, pulse
                        FROM signal s1
                        WHERE s1.target = types.name
                          AND ((s1.press < signal.press) OR
                               (s1.press = signal.press AND s1.round < signal.round) OR
                               (s1.press = signal.press AND s1.round = signal.round AND s1.source <= signal.source))
                        OPTIONS (DISTINCT ON INPUT GROUP SIZE = 1000)
                        ORDER BY source, press DESC, round DESC
                    )
                    WHERE pulse = 'hi'))
                 THEN 'lo'
                 ELSE 'hi'
            END
        FROM signal, types
        WHERE signal.target = types.name
            AND types.op = '&'
    ),

    -- A record of a pulse into an operator, from another operator.
    -- We track the source so that '&' operators can make any sense.
    signal(source TEXT, target TEXT, press INT, round INT, pulse TEXT) AS (
        SELECT pulses.name, links.link, pulses.press, pulses.round, pulses.pulse
        FROM pulses, links
        WHERE pulses.name = links.name
          AND pulses.round > 0
    ),

    part1(pulse TEXT, count BIGINT) AS (
        SELECT pulse, count(*) FROM signal GROUP BY pulse
    ),

    potato(x INT) AS (SELECT 1)

SELECT * FROM signal WHERE target = 'cn' AND pulse = 'hi';
----
Explained Query:
  With
    cte l0 =
      Project (#1) // { arity: 1 }
        FlatMap unnest_array(regexp_split_to_array["\n", case_insensitive=false](#0{input})) // { arity: 2 }
          ReadStorage materialize.public.input // { arity: 1 }
    cte l1 =
      Project (#3, #4) // { arity: 2 }
        Map (regexp_split_to_array[" ", case_insensitive=false](#0{line}), substr(array_index(#2, 1), 2), btrim(array_index(#2, integer_to_bigint(#1{x})), ",")) // { arity: 5 }
          FlatMap generate_series(3, (regexp_split_to_array[" ", case_insensitive=false](#0{line}) array_length 1), 1) // { arity: 2 }
            Get l0 // { arity: 1 }
    cte l2 =
      Project (#1, #2) // { arity: 2 }
        Map (array_index(regexp_split_to_array[" ", case_insensitive=false](#0{line}), 1), substr(#1, 2)) // { arity: 3 }
          Get l0 // { arity: 1 }
  Return // { arity: 5 }
    With Mutually Recursive
      cte l3 =
        Distinct project=[#0, #1] monotonic // { arity: 2 }
          Union // { arity: 2 }
            Project (#0, #2) // { arity: 2 }
              Filter (#1{counter} > 0) // { arity: 3 }
                Map ((#1{counter} - 1)) // { arity: 3 }
                  Get l3 // { arity: 2 }
            Project (#2, #3) // { arity: 2 }
              Filter (#1{counter} = 0) AND (#0{press} < 4100) // { arity: 4 }
                Map ((#0{press} + 1), 20) // { arity: 4 }
                  Get l3 // { arity: 2 }
            Constant // { arity: 2 }
              - (1, 1)
      cte l4 =
        Union // { arity: 4 }
          Project (#2, #0, #3, #4) // { arity: 4 }
            Filter (#1{counter} = 0) // { arity: 5 }
              Map ("roadcaster", 1, "lo") // { arity: 5 }
                Get l3 // { arity: 2 }
          Filter (#2{round} > 0) // { arity: 4 }
            Get l12 // { arity: 4 }
          Filter (#2{round} > 0) // { arity: 4 }
            Get l26 // { arity: 4 }
      cte l5 =
        ArrangeBy keys=[[#1{target}]] // { arity: 4 }
          Project (#0..=#3) // { arity: 4 }
            Filter (#4{pulse} = "lo") AND (#1{target}) IS NOT NULL // { arity: 5 }
              Get l27 // { arity: 5 }
      cte l6 =
        Project (#0..=#3, #5) // { arity: 5 }
          Map ((#3{round} + 1)) // { arity: 6 }
            Join on=(#1{target} = #4{name}) type=differential // { arity: 5 }
              implementation
                %0:l5[#1{target}]Kef » %1:l2[#0{name}]Kef
              Get l5 // { arity: 4 }
              ArrangeBy keys=[[#0{name}]] // { arity: 1 }
                Project (#1) // { arity: 1 }
                  Filter ("%" = substr(#0, 1, 1)) // { arity: 2 }
                    Get l2 // { arity: 2 }
      cte l7 =
        Distinct project=[#0, #2, #3, #1] // { arity: 4 }
          Project (#0..=#3) // { arity: 4 }
            Get l6 // { arity: 5 }
      cte l8 =
        Reduce group_by=[#0..=#3] aggregates=[count(*)] // { arity: 5 }
          Project (#0..=#3) // { arity: 4 }
            Filter ((#6{press} < #1{press}) OR ((#1{press} = #6{press}) AND ((#7{round} < #2{round}) OR ((#2{round} = #7{round}) AND (#4{source} < #0{source}))))) // { arity: 8 }
              Join on=(#3{name} = #5{target}) type=differential // { arity: 8 }
                implementation
                  %1:l5[#1{target}]Kef » %0:l7[#3{name}]Kef
                ArrangeBy keys=[[#3{name}]] // { arity: 4 }
                  Get l7 // { arity: 4 }
                Get l5 // { arity: 4 }
      cte l9 =
        ArrangeBy keys=[[#0..=#3]] // { arity: 4 }
          Get l7 // { arity: 4 }
      cte l10 =
        Union // { arity: 5 }
          Get l8 // { arity: 5 }
          Project (#0..=#3, #8) // { arity: 5 }
            Map (0) // { arity: 9 }
              Join on=(#0 = #4 AND #1 = #5 AND #2 = #6 AND #3 = #7) type=differential // { arity: 8 }
                implementation
                  %1:l9[#0..=#3]UKKKK » %0[#0..=#3]KKKK
                ArrangeBy keys=[[#0..=#3]] // { arity: 4 }
                  Union // { arity: 4 }
                    Negate // { arity: 4 }
                      Project (#0..=#3) // { arity: 4 }
                        Get l8 // { arity: 5 }
                    Get l7 // { arity: 4 }
                Get l9 // { arity: 4 }
      cte l11 =
        Union // { arity: 5 }
          Get l10 // { arity: 5 }
          Project (#0..=#3, #5) // { arity: 5 }
            FlatMap guard_subquery_size(#4{count}) // { arity: 6 }
              Reduce group_by=[#0..=#3] aggregates=[count(*)] // { arity: 5 }
                Project (#0..=#3) // { arity: 4 }
                  Get l10 // { arity: 5 }
      cte l12 =
        Project (#1, #2, #4, #10) // { arity: 4 }
          Map (case when (0 = (#9{count} % 2)) then "hi" else "lo" end) // { arity: 11 }
            Join on=(#0 = #5 AND #1 = #8 AND #2 = #6 AND #3 = #7) type=differential // { arity: 10 }
              implementation
                %0:l6[#0, #2, #3, #1]KKKK » %1[#0..=#3]KKKK
              ArrangeBy keys=[[#0, #2, #3, #1]] // { arity: 5 }
                Get l6 // { arity: 5 }
              ArrangeBy keys=[[#0..=#3]] // { arity: 5 }
                Union // { arity: 5 }
                  Get l11 // { arity: 5 }
                  Project (#0..=#3, #8) // { arity: 5 }
                    Map (null) // { arity: 9 }
                      Join on=(#0 = #4 AND #1 = #5 AND #2 = #6 AND #3 = #7) type=differential // { arity: 8 }
                        implementation
                          %1:l9[#0..=#3]UKKKK » %0[#0..=#3]KKKK
                        ArrangeBy keys=[[#0..=#3]] // { arity: 4 }
                          Union // { arity: 4 }
                            Negate // { arity: 4 }
                              Distinct project=[#0..=#3] // { arity: 4 }
                                Project (#0..=#3) // { arity: 4 }
                                  Get l11 // { arity: 5 }
                            Get l7 // { arity: 4 }
                        Get l9 // { arity: 4 }
      cte l13 =
        Filter (#1{target}) IS NOT NULL // { arity: 5 }
          Get l27 // { arity: 5 }
      cte l14 =
        Project (#0..=#3, #5) // { arity: 5 }
          Map ((#3{round} + 1)) // { arity: 6 }
            Join on=(#1{target} = #4{name}) type=differential // { arity: 5 }
              implementation
                %1:l2[#0{name}]Kef » %0:l13[#1{target}]Kef
              ArrangeBy keys=[[#1{target}]] // { arity: 4 }
                Project (#0..=#3) // { arity: 4 }
                  Get l13 // { arity: 5 }
              ArrangeBy keys=[[#0{name}]] // { arity: 1 }
                Project (#1) // { arity: 1 }
                  Filter ("&" = substr(#0, 1, 1)) // { arity: 2 }
                    Get l2 // { arity: 2 }
      cte l15 =
        Distinct project=[#0] // { arity: 1 }
          Project (#1) // { arity: 1 }
            Get l14 // { arity: 5 }
      cte l16 =
        ArrangeBy keys=[[#0{name}]] // { arity: 1 }
          Get l15 // { arity: 1 }
      cte l17 =
        Reduce group_by=[#0] aggregates=[count(*)] // { arity: 2 }
          Project (#0) // { arity: 1 }
            Join on=(#0{name} = #1{link}) type=differential // { arity: 2 }
              implementation
                %0:l16[#0{name}]UK » %1:l1[#0{link}]K
              Get l16 // { arity: 1 }
              ArrangeBy keys=[[#0{link}]] // { arity: 1 }
                Project (#1) // { arity: 1 }
                  Filter (#1{link}) IS NOT NULL // { arity: 2 }
                    Get l1 // { arity: 2 }
      cte l18 =
        Union // { arity: 2 }
          Get l17 // { arity: 2 }
          Project (#0, #2) // { arity: 2 }
            Map (0) // { arity: 3 }
              Join on=(#0 = #1) type=differential // { arity: 2 }
                implementation
                  %1:l16[#0]UK » %0[#0]K
                ArrangeBy keys=[[#0]] // { arity: 1 }
                  Union // { arity: 1 }
                    Negate // { arity: 1 }
                      Project (#0) // { arity: 1 }
                        Get l17 // { arity: 2 }
                    Get l15 // { arity: 1 }
                Get l16 // { arity: 1 }
      cte l19 =
        Union // { arity: 2 }
          Get l18 // { arity: 2 }
          Project (#0, #2) // { arity: 2 }
            FlatMap guard_subquery_size(#1{count}) // { arity: 3 }
              Reduce group_by=[#0] aggregates=[count(*)] // { arity: 2 }
                Project (#0) // { arity: 1 }
                  Get l18 // { arity: 2 }
      cte l20 =
        Project (#0..=#4, #6{count}) // { arity: 6 }
          Join on=(#1 = #5) type=differential // { arity: 7 }
            implementation
              %0:l14[#1]K » %1[#0]K
            ArrangeBy keys=[[#1]] // { arity: 5 }
              Get l14 // { arity: 5 }
            ArrangeBy keys=[[#0]] // { arity: 2 }
              Union // { arity: 2 }
                Get l19 // { arity: 2 }
                Project (#0, #2) // { arity: 2 }
                  Map (null) // { arity: 3 }
                    Join on=(#0 = #1) type=differential // { arity: 2 }
                      implementation
                        %1:l16[#0]UK » %0[#0]K
                      ArrangeBy keys=[[#0]] // { arity: 1 }
                        Union // { arity: 1 }
                          Negate // { arity: 1 }
                            Distinct project=[#0] // { arity: 1 }
                              Project (#0) // { arity: 1 }
                                Get l19 // { arity: 2 }
                          Get l15 // { arity: 1 }
                      Get l16 // { arity: 1 }
      cte l21 =
        Distinct project=[#0, #2, #3, #1] // { arity: 4 }
          Project (#0..=#3) // { arity: 4 }
            Get l20 // { arity: 6 }
      cte l22 =
        Reduce group_by=[#0..=#3] aggregates=[count(*)] // { arity: 5 }
          Project (#0..=#3) // { arity: 4 }
            Filter (#7{pulse} = "hi") // { arity: 8 }
              TopK group_by=[#0, #1, #2, #3, #4] order_by=[#5 desc nulls_first, #6 desc nulls_first] limit=1 exp_group_size=1000 // { arity: 8 }
                Project (#0..=#4, #6..=#8) // { arity: 8 }
                  Filter ((#6{press} < #1{press}) OR ((#1{press} = #6{press}) AND ((#7{round} < #2{round}) OR ((#2{round} = #7{round}) AND (#4{source} <= #0{source}))))) // { arity: 9 }
                    Join on=(#3{name} = #5{target}) type=differential // { arity: 9 }
                      implementation
                        %0:l21[#3{name}]K » %1:l13[#1{target}]K
                      ArrangeBy keys=[[#3{name}]] // { arity: 4 }
                        Get l21 // { arity: 4 }
                      ArrangeBy keys=[[#1{target}]] // { arity: 5 }
                        Get l13 // { arity: 5 }
      cte l23 =
        ArrangeBy keys=[[#0..=#3]] // { arity: 4 }
          Get l21 // { arity: 4 }
      cte l24 =
        Union // { arity: 5 }
          Get l22 // { arity: 5 }
          Project (#0..=#3, #8) // { arity: 5 }
            Map (0) // { arity: 9 }
              Join on=(#0 = #4 AND #1 = #5 AND #2 = #6 AND #3 = #7) type=differential // { arity: 8 }
                implementation
                  %1:l23[#0..=#3]UKKKK » %0[#0..=#3]KKKK
                ArrangeBy keys=[[#0..=#3]] // { arity: 4 }
                  Union // { arity: 4 }
                    Negate // { arity: 4 }
                      Project (#0..=#3) // { arity: 4 }
                        Get l22 // { arity: 5 }
                    Get l21 // { arity: 4 }
                Get l23 // { arity: 4 }
      cte l25 =
        Union // { arity: 5 }
          Get l24 // { arity: 5 }
          Project (#0..=#3, #5) // { arity: 5 }
            FlatMap guard_subquery_size(#4{count}) // { arity: 6 }
              Reduce group_by=[#0..=#3] aggregates=[count(*)] // { arity: 5 }
                Project (#0..=#3) // { arity: 4 }
                  Get l24 // { arity: 5 }
      cte l26 =
        Project (#1, #2, #4, #11) // { arity: 4 }
          Map (case when (#5{count} = #10{count}) then "lo" else "hi" end) // { arity: 12 }
            Join on=(#0 = #6 AND #1 = #9 AND #2 = #7 AND #3 = #8) type=differential // { arity: 11 }
              implementation
                %0:l20[#0, #2, #3, #1]KKKK » %1[#0..=#3]KKKK
              ArrangeBy keys=[[#0, #2, #3, #1]] // { arity: 6 }
                Get l20 // { arity: 6 }
              ArrangeBy keys=[[#0..=#3]] // { arity: 5 }
                Union // { arity: 5 }
                  Get l25 // { arity: 5 }
                  Project (#0..=#3, #8) // { arity: 5 }
                    Map (null) // { arity: 9 }
                      Join on=(#0 = #4 AND #1 = #5 AND #2 = #6 AND #3 = #7) type=differential // { arity: 8 }
                        implementation
                          %1:l23[#0..=#3]UKKKK » %0[#0..=#3]KKKK
                        ArrangeBy keys=[[#0..=#3]] // { arity: 4 }
                          Union // { arity: 4 }
                            Negate // { arity: 4 }
                              Distinct project=[#0..=#3] // { arity: 4 }
                                Project (#0..=#3) // { arity: 4 }
                                  Get l25 // { arity: 5 }
                            Get l21 // { arity: 4 }
                        Get l23 // { arity: 4 }
      cte l27 =
        Project (#0, #5, #1..=#3) // { arity: 5 }
          Join on=(#0{name} = #4{name}) type=differential // { arity: 6 }
            implementation
              %0:l4[#0{name}]K » %1:l1[#0{name}]K
            ArrangeBy keys=[[#0{name}]] // { arity: 4 }
              Get l4 // { arity: 4 }
            ArrangeBy keys=[[#0{name}]] // { arity: 2 }
              Filter (#0{name}) IS NOT NULL // { arity: 2 }
                Get l1 // { arity: 2 }
    Return // { arity: 5 }
      Filter (#1{target} = "cn") AND (#4{pulse} = "hi") // { arity: 5 }
        Get l27 // { arity: 5 }

Source materialize.public.input

Target cluster: quickstart

EOF