# 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_1223.md mode cockroach statement ok CREATE TABLE input (input TEXT); statement ok INSERT INTO input VALUES ( '##############.################################################################### #....#.#...#....>..#...######..#.#..#.#.#.#..#..#...#.#...###..##.###...##........ #####..#.v##....v...#....#.#....#.###......<..#...#....#>..#.........#.##...#..... .##..#.##...#..........#...#...#..#.#..^#..#...#.#..#.#.##<.#.<.##.#..#.#......### ..#.##.#.##..v..................#...#.##.#.##.>#.#........#....##.....#..#...#.... #..<#.##..............#..#.......^##.#.#.......v....##...###.#.##.<..#....#....... .#....#.#^..#.......####.#^........>........##.........###..#.#.#...#...#...##.... ##......#.#......#..#.#.##^..#....^...###.#..#.#.<.........<...##..#.#....<.#.##.# #.#.#...................>....#..##....#....##......#...^....#.#....#.#..v.##.#..## ###.#...#..##.#.#.......#....v#.###.............#...##.###......#.#.#.#.....#..#^. ##.#..........#.#.#.#...#....#......#.....#.#....##...##.....^.###.#.......###.<.. ...........#..#...#^.#..........#..##..##..#....###...>#.<#..>.##..#......##.#.... .>..##.#.....#.....####..#.#.#............##><.....#...###..#.....###.##.........# ...#.....>....#.v>^#...##......#..###..#...##...^###.#........#..#...#.#..#.....#> ^..^...........#.#.#.#.##.##.....#.....#.##....#.##............#.##...#..#......^. .#........#.<#.....v.......#...#v......#.#........#...#..#.#...#>.###....#v....##. .##.....#v....#^#v#.#.........#^#......<...#..##...#.#.v.###....##..#...#.#......# .#..#.##..#.##..#>..#.......<......##..#.##..###.##...#...#.#...#.#....#..##..#..# .v.#.#....>..............#.####.#...>......v#.####..#......v....##^.##.#...#...^## ##############.###################################################################'); query IIII WITH MUTUALLY RECURSIVE lines(r INT, line TEXT) AS ( SELECT r, regexp_split_to_array(input, '\n')[r] as line FROM input, generate_series(1, array_length(regexp_split_to_array(input, '\n'), 1)) r ), cells(r INT, c INT, symbol TEXT) AS ( SELECT r, c, substring(line, c, 1) FROM lines, generate_series(1, length(line)) c ), -- Part one: longest path (on probably a DAG) paths(r INT, c INT) AS ( SELECT r, c FROM cells WHERE symbol = '.' ), steps(r1 INT, c1 INT, r2 INT, c2 INT) AS ( SELECT r, c, r + 1, c FROM paths WHERE (r + 1, c) IN (SELECT * FROM PATHS) UNION SELECT r, c, r - 1, c FROM paths WHERE (r - 1, c) IN (SELECT * FROM PATHS) UNION SELECT r, c, r, c + 1 FROM paths WHERE (r, c + 1) IN (SELECT * FROM PATHS) UNION SELECT r, c, r, c - 1 FROM paths WHERE (r, c - 1) IN (SELECT * FROM PATHS) ), -- A directional trip, forced by a slope and the no-revisting rule. force(r1 INT, c1 INT, r2 INT, c2 INT) AS ( SELECT r-1, c, r+1, c FROM cells WHERE symbol = 'v' UNION ALL SELECT r+1, c, r-1, c FROM cells WHERE symbol = '^' UNION ALL SELECT r, c-1, r, c+1 FROM cells WHERE symbol = '>' UNION ALL SELECT r, c+1, r, c-1 FROM cells WHERE symbol = '<' ), dists(r INT, c INT, d INT) AS ( SELECT 1, 2, 0 UNION SELECT steps.r2, steps.c2, 1 + MIN(d) FROM dists, steps WHERE dists.r = steps.r1 AND dists.c = steps.c1 GROUP BY steps.r2, steps.c2 UNION SELECT force.r2, force.c2, 2 + MAX(d) FROM dists, force WHERE dists.r = force.r1 AND dists.c = force.c1 GROUP BY force.r2, force.c2 ), -- Part two: longest path on definitely not a DAG. -- There are 32 optional nodes (not including first and last nodes) -- Clearly meant to pack in to an int and avoid duplication. paths2(r INT, c INT) AS ( SELECT r, c FROM cells WHERE symbol != '#' ), steps2(r1 INT, c1 INT, r2 INT, c2 INT) AS ( SELECT r, c, r + 1, c FROM paths2 WHERE (r + 1, c) IN (SELECT * FROM paths2) UNION SELECT r, c, r - 1, c FROM paths2 WHERE (r - 1, c) IN (SELECT * FROM paths2) UNION SELECT r, c, r, c + 1 FROM paths2 WHERE (r, c + 1) IN (SELECT * FROM paths2) UNION SELECT r, c, r, c - 1 FROM paths2 WHERE (r, c - 1) IN (SELECT * FROM paths2) ), -- Locations where a choice exists (or start/end). nodes(r INT, c INT) AS ( SELECT r1, c1 FROM steps2 GROUP BY r1, c1 HAVING COUNT(*) != 2 ), -- Determine node-to-node path lengths. Do not cross nodes. trail(r1 INT, c1 INT, d INT, r2 INT, c2 INT) AS ( SELECT r1, c1, MIN(d), r2, c2 FROM ( SELECT r1, c1, 1 d, r2, c2 FROM steps2 WHERE (r1, c1) IN (SELECT * FROM nodes) UNION ALL SELECT trail.r1, trail.c1, d + 1, steps2.r2, steps2.c2 FROM trail, steps2 WHERE trail.r2 = steps2.r1 AND trail.c2 = steps2.c1 AND (trail.r1 != steps2.r2 OR trail.c1 != steps2.c2) AND (steps2.r1, steps2.c1) NOT IN (SELECT * FROM nodes) ) GROUP BY r1, c1, r2, c2 ), links(r1 INT, c1 INT, d INT, r2 INT, c2 INT) AS ( SELECT * FROM trail WHERE (r2, c2) IN (SELECT * FROM nodes) ), -- These rows in links show that (12, 20) and (130, 126) are mandatory, -- and are the first moments we have a choice. The remaining 32 nodes -- can each get a number, and be used in a bit pattern somewhere. -- -- 1 | 2 | 105 | 12 | 20 -- 141 | 140 | 121 | 130 | 126 -- Re-key nodes to dense integers. internal(r INT, c INT, id INT) AS ( SELECT r, c, ( SELECT COUNT(*) FROM nodes n1 WHERE (n1.r < n2.r OR (n1.r = n2.r AND n1.c < n2.c)) AND (n1.r, n1.c) NOT IN (VALUES (1,2), (12,20), (130,126), (141,140)) ) FROM nodes n2 WHERE (r, c) NOT IN (VALUES (1,2), (12,20), (130,126), (141,140)) ), longest(r INT, c INT, d INT, v BIGINT) AS ( SELECT r, c, MAX(d), v FROM ( SELECT 12 r, 20 c, 0 d, 0 v UNION ALL SELECT r2, c2, longest.d + links.d, v + (1::BIGINT << internal.id) FROM longest, links, internal WHERE longest.r = links.r1 AND longest.c = links.c1 AND links.r2 = internal.r AND links.c2 = internal.c AND ((v >> internal.id) % 2) != 1 ) GROUP BY r, c, v ), potato(x INT) AS ( SELECT 1 ) SELECT * FROM longest ORDER BY d DESC; ---- 12 20 0 0 query T multiline EXPLAIN OPTIMIZED PLAN WITH(humanized expressions, arity, join implementations) AS VERBOSE TEXT FOR WITH MUTUALLY RECURSIVE lines(r INT, line TEXT) AS ( SELECT r, regexp_split_to_array(input, '\n')[r] as line FROM input, generate_series(1, array_length(regexp_split_to_array(input, '\n'), 1)) r ), cells(r INT, c INT, symbol TEXT) AS ( SELECT r, c, substring(line, c, 1) FROM lines, generate_series(1, length(line)) c ), -- Part one: longest path (on probably a DAG) paths(r INT, c INT) AS ( SELECT r, c FROM cells WHERE symbol = '.' ), steps(r1 INT, c1 INT, r2 INT, c2 INT) AS ( SELECT r, c, r + 1, c FROM paths WHERE (r + 1, c) IN (SELECT * FROM PATHS) UNION SELECT r, c, r - 1, c FROM paths WHERE (r - 1, c) IN (SELECT * FROM PATHS) UNION SELECT r, c, r, c + 1 FROM paths WHERE (r, c + 1) IN (SELECT * FROM PATHS) UNION SELECT r, c, r, c - 1 FROM paths WHERE (r, c - 1) IN (SELECT * FROM PATHS) ), -- A directional trip, forced by a slope and the no-revisting rule. force(r1 INT, c1 INT, r2 INT, c2 INT) AS ( SELECT r-1, c, r+1, c FROM cells WHERE symbol = 'v' UNION ALL SELECT r+1, c, r-1, c FROM cells WHERE symbol = '^' UNION ALL SELECT r, c-1, r, c+1 FROM cells WHERE symbol = '>' UNION ALL SELECT r, c+1, r, c-1 FROM cells WHERE symbol = '<' ), dists(r INT, c INT, d INT) AS ( SELECT 1, 2, 0 UNION SELECT steps.r2, steps.c2, 1 + MIN(d) FROM dists, steps WHERE dists.r = steps.r1 AND dists.c = steps.c1 GROUP BY steps.r2, steps.c2 UNION SELECT force.r2, force.c2, 2 + MAX(d) FROM dists, force WHERE dists.r = force.r1 AND dists.c = force.c1 GROUP BY force.r2, force.c2 ), -- Part two: longest path on definitely not a DAG. -- There are 32 optional nodes (not including first and last nodes) -- Clearly meant to pack in to an int and avoid duplication. paths2(r INT, c INT) AS ( SELECT r, c FROM cells WHERE symbol != '#' ), steps2(r1 INT, c1 INT, r2 INT, c2 INT) AS ( SELECT r, c, r + 1, c FROM paths2 WHERE (r + 1, c) IN (SELECT * FROM paths2) UNION SELECT r, c, r - 1, c FROM paths2 WHERE (r - 1, c) IN (SELECT * FROM paths2) UNION SELECT r, c, r, c + 1 FROM paths2 WHERE (r, c + 1) IN (SELECT * FROM paths2) UNION SELECT r, c, r, c - 1 FROM paths2 WHERE (r, c - 1) IN (SELECT * FROM paths2) ), -- Locations where a choice exists (or start/end). nodes(r INT, c INT) AS ( SELECT r1, c1 FROM steps2 GROUP BY r1, c1 HAVING COUNT(*) != 2 ), -- Determine node-to-node path lengths. Do not cross nodes. trail(r1 INT, c1 INT, d INT, r2 INT, c2 INT) AS ( SELECT r1, c1, MIN(d), r2, c2 FROM ( SELECT r1, c1, 1 d, r2, c2 FROM steps2 WHERE (r1, c1) IN (SELECT * FROM nodes) UNION ALL SELECT trail.r1, trail.c1, d + 1, steps2.r2, steps2.c2 FROM trail, steps2 WHERE trail.r2 = steps2.r1 AND trail.c2 = steps2.c1 AND (trail.r1 != steps2.r2 OR trail.c1 != steps2.c2) AND (steps2.r1, steps2.c1) NOT IN (SELECT * FROM nodes) ) GROUP BY r1, c1, r2, c2 ), links(r1 INT, c1 INT, d INT, r2 INT, c2 INT) AS ( SELECT * FROM trail WHERE (r2, c2) IN (SELECT * FROM nodes) ), -- These rows in links show that (12, 20) and (130, 126) are mandatory, -- and are the first moments we have a choice. The remainaing 32 nodes -- can each get a number, and be used in a bit pattern somewhere. -- -- 1 | 2 | 105 | 12 | 20 -- 141 | 140 | 121 | 130 | 126 -- Re-key nodes to dense integers. internal(r INT, c INT, id INT) AS ( SELECT r, c, ( SELECT COUNT(*) FROM nodes n1 WHERE (n1.r < n2.r OR (n1.r = n2.r AND n1.c < n2.c)) AND (n1.r, n1.c) NOT IN (VALUES (1,2), (12,20), (130,126), (141,140)) ) FROM nodes n2 WHERE (r, c) NOT IN (VALUES (1,2), (12,20), (130,126), (141,140)) ), longest(r INT, c INT, d INT, v BIGINT) AS ( SELECT r, c, MAX(d), v FROM ( SELECT 12 r, 20 c, 0 d, 0 v UNION ALL SELECT r2, c2, longest.d + links.d, v + (1::BIGINT << internal.id) FROM longest, links, internal WHERE longest.r = links.r1 AND longest.c = links.c1 AND links.r2 = internal.r AND links.c2 = internal.c AND ((v >> internal.id) % 2) != 1 ) GROUP BY r, c, v ), potato(x INT) AS ( SELECT 1 ) SELECT * FROM longest ORDER BY d DESC; ---- Explained Query: Finish order_by=[#2{max} desc nulls_first] output=[#0..=#3] With cte l0 = Project (#0, #2) // { arity: 2 } Filter ("#" != substr(#1{line}, #2{c}, 1)) // { arity: 3 } FlatMap generate_series(1, char_length(#1{line}), 1) // { arity: 3 } Project (#1, #2) // { arity: 2 } Map (array_index(regexp_split_to_array["\n", case_insensitive=false](#0{input}), integer_to_bigint(#1{r}))) // { arity: 3 } FlatMap generate_series(1, (regexp_split_to_array["\n", case_insensitive=false](#0{input}) array_length 1), 1) // { arity: 2 } ReadStorage materialize.public.input // { arity: 1 } cte l1 = Distinct project=[#0, #1] // { arity: 2 } Get l0 // { arity: 2 } cte l2 = ArrangeBy keys=[[#0, #1]] // { arity: 2 } Get l0 // { arity: 2 } cte l3 = Distinct project=[#0..=#3] // { arity: 4 } Union // { arity: 4 } Project (#0..=#2, #1) // { arity: 4 } Distinct project=[#0..=#2] // { arity: 3 } Union // { arity: 3 } Project (#0, #1, #4) // { arity: 3 } Map ((#0{r} + 1)) // { arity: 5 } Join on=(#0 = #2 AND #1 = #3) type=differential // { arity: 4 } implementation %1[#0, #1]UKKA » %0:l2[#0, #1]KK Get l2 // { arity: 2 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Distinct project=[#0, #1] // { arity: 2 } Project (#0, #1) // { arity: 2 } Join on=(#1{c} = #3{right_col1_13} AND #2{right_col0_12} = (#0{r} + 1)) type=differential // { arity: 4 } implementation %0:l1[(#0{r} + 1), #1{c}]KK » %1:l2[#0{right_col0_12}, #1{right_col1_13}]KK ArrangeBy keys=[[(#0{r} + 1), #1{c}]] // { arity: 2 } Get l1 // { arity: 2 } Get l2 // { arity: 2 } Project (#0, #1, #4) // { arity: 3 } Map ((#0{r} - 1)) // { arity: 5 } Join on=(#0 = #2 AND #1 = #3) type=differential // { arity: 4 } implementation %1[#0, #1]UKKA » %0:l2[#0, #1]KK Get l2 // { arity: 2 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Distinct project=[#0, #1] // { arity: 2 } Project (#0, #1) // { arity: 2 } Join on=(#1{c} = #3{right_col1_16} AND #2{right_col0_15} = (#0{r} - 1)) type=differential // { arity: 4 } implementation %0:l1[(#0{r} - 1), #1{c}]KK » %1:l2[#0{right_col0_15}, #1{right_col1_16}]KK ArrangeBy keys=[[(#0{r} - 1), #1{c}]] // { arity: 2 } Get l1 // { arity: 2 } Get l2 // { arity: 2 } Project (#0, #1, #0, #4) // { arity: 4 } Map ((#1{c} + 1)) // { arity: 5 } Join on=(#0 = #2 AND #1 = #3) type=differential // { arity: 4 } implementation %1[#0, #1]UKKA » %0:l2[#0, #1]KK Get l2 // { arity: 2 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Distinct project=[#0, #1] // { arity: 2 } Project (#0, #1) // { arity: 2 } Join on=(#0{r} = #2{right_col0_18} AND #3{right_col1_19} = (#1{c} + 1)) type=differential // { arity: 4 } implementation %0:l1[#0{r}, (#1{c} + 1)]KK » %1:l2[#0{right_col0_18}, #1{right_col1_19}]KK ArrangeBy keys=[[#0{r}, (#1{c} + 1)]] // { arity: 2 } Get l1 // { arity: 2 } Get l2 // { arity: 2 } Project (#0, #1, #0, #4) // { arity: 4 } Map ((#1{c} - 1)) // { arity: 5 } Join on=(#0 = #2 AND #1 = #3) type=differential // { arity: 4 } implementation %1[#0, #1]UKKA » %0:l2[#0, #1]KK Get l2 // { arity: 2 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Distinct project=[#0, #1] // { arity: 2 } Project (#0, #1) // { arity: 2 } Join on=(#0{r} = #2{right_col0_21} AND #3{right_col1_22} = (#1{c} - 1)) type=differential // { arity: 4 } implementation %0:l1[#0{r}, (#1{c} - 1)]KK » %1:l2[#0{right_col0_21}, #1{right_col1_22}]KK ArrangeBy keys=[[#0{r}, (#1{c} - 1)]] // { arity: 2 } Get l1 // { arity: 2 } Get l2 // { arity: 2 } cte l4 = Project (#0, #1) // { arity: 2 } Filter (#2{count} != 2) // { arity: 3 } Reduce group_by=[#0, #1] aggregates=[count(*)] // { arity: 3 } Project (#0, #1) // { arity: 2 } Get l3 // { arity: 4 } cte l5 = Union // { arity: 2 } Negate // { arity: 2 } Distinct project=[#0, #1] // { arity: 2 } Project (#0, #1) // { arity: 2 } Filter (#0{r} = #2{right_col0_36}) AND (#1{c} = #3{right_col1_37}) // { arity: 4 } FlatMap wrap2(1, 2, 12, 20, 130, 126, 141, 140) // { arity: 4 } Get l4 // { arity: 2 } Get l4 // { arity: 2 } cte l6 = Distinct project=[#0, #1] // { arity: 2 } Get l5 // { arity: 2 } cte l7 = Filter ((#2{r} < #0{r}) OR ((#0{r} = #2{r}) AND (#3{c} < #1{c}))) // { arity: 4 } CrossJoin type=differential // { arity: 4 } implementation %0:l6[×] » %1:l4[×] ArrangeBy keys=[[]] // { arity: 2 } Get l6 // { arity: 2 } ArrangeBy keys=[[]] // { arity: 2 } Get l4 // { arity: 2 } cte l8 = Distinct project=[#0, #1] // { arity: 2 } Project (#2, #3) // { arity: 2 } Get l7 // { arity: 4 } cte l9 = Reduce group_by=[#0, #1] aggregates=[count(*)] // { arity: 3 } Project (#0, #1) // { arity: 2 } Join on=(#2 = #4 AND #3 = #5) type=differential // { arity: 6 } implementation %0:l7[#2, #3]KK » %1[#0, #1]KK ArrangeBy keys=[[#2, #3]] // { arity: 4 } Get l7 // { arity: 4 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Union // { arity: 2 } Negate // { arity: 2 } Distinct project=[#0, #1] // { arity: 2 } Project (#0, #1) // { arity: 2 } Filter (#0{r} = #2{right_col0_33}) AND (#1{c} = #3{right_col1_34}) // { arity: 4 } FlatMap wrap2(1, 2, 12, 20, 130, 126, 141, 140) // { arity: 4 } Get l8 // { arity: 2 } Get l8 // { arity: 2 } cte l10 = ArrangeBy keys=[[#0{r1}, #1{c1}]] // { arity: 4 } Get l3 // { arity: 4 } cte l11 = ArrangeBy keys=[[#0, #1]] // { arity: 2 } Get l4 // { arity: 2 } Return // { arity: 4 } With Mutually Recursive cte l12 = Project (#0..=#4, #7, #8) // { arity: 7 } Filter ((#0{r1} != #7{r2}) OR (#1{c1} != #8{c2})) // { arity: 9 } Join on=(#3{r2} = #5{r1} AND #4{c2} = #6{c1}) type=differential // { arity: 9 } implementation %0:l14[#3{r2}, #4{c2}]KK » %1:l10[#0{r1}, #1{c1}]KK ArrangeBy keys=[[#3{r2}, #4{c2}]] // { arity: 5 } Get l14 // { arity: 5 } Get l10 // { arity: 4 } cte l13 = Distinct project=[#0, #1] // { arity: 2 } Project (#3, #4) // { arity: 2 } Get l12 // { arity: 7 } cte l14 = Project (#0, #1, #4{min}, #2, #3) // { arity: 5 } Reduce group_by=[#0, #1, #3, #4] aggregates=[min(#2{d})] // { arity: 5 } Union // { arity: 5 } Project (#0, #1, #6, #2, #3) // { arity: 5 } Map (1) // { arity: 7 } Join on=(#0 = #4 AND #1 = #5) type=differential // { arity: 6 } implementation %1:l11[#0, #1]UKK » %0:l10[#0, #1]KK Get l10 // { arity: 4 } Get l11 // { arity: 2 } Project (#0, #1, #9, #5, #6) // { arity: 5 } Map ((#2{d} + 1)) // { arity: 10 } Join on=(#3 = #7 AND #4 = #8) type=differential // { arity: 9 } implementation %0:l12[#3, #4]KK » %1[#0, #1]KK ArrangeBy keys=[[#3, #4]] // { arity: 7 } Get l12 // { arity: 7 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Union // { arity: 2 } Negate // { arity: 2 } Project (#0, #1) // { arity: 2 } Join on=(#0 = #2 AND #1 = #3) type=differential // { arity: 4 } implementation %0:l13[#0, #1]UKK » %1:l11[#0, #1]UKK ArrangeBy keys=[[#0, #1]] // { arity: 2 } Get l13 // { arity: 2 } Get l11 // { arity: 2 } Get l13 // { arity: 2 } cte l15 = Project (#0, #1, #3{max}, #2) // { arity: 4 } Reduce group_by=[#0, #1, #3] aggregates=[max(#2{d})] // { arity: 4 } Union // { arity: 4 } Project (#7, #8, #15, #16) // { arity: 4 } Filter (1 != ((#3{v} >> #14) % 2)) // { arity: 17 } Map (bigint_to_integer(#13{count}), (#2{d} + #6{min}), (#3{v} + (1 << #14{id}))) // { arity: 17 } Join on=(#0{r} = #4{r1} AND #1{c} = #5{c1} AND #7 = #9 = #11 AND #8 = #10 = #12) type=delta // { arity: 14 } implementation %0:l15 » %1:l14[#0{r1}, #1{c1}]KK » %3[#0, #1]UKK » %2:l5[#0, #1]KK %1:l14 » %3[#0, #1]UKK » %0:l15[#0{r}, #1{c}]KK » %2:l5[#0, #1]KK %2:l5 » %3[#0, #1]UKK » %1:l14[#3, #4]KK » %0:l15[#0{r}, #1{c}]KK %3 » %1:l14[#3, #4]KK » %0:l15[#0{r}, #1{c}]KK » %2:l5[#0, #1]KK ArrangeBy keys=[[#0{r}, #1{c}]] // { arity: 4 } Get l15 // { arity: 4 } ArrangeBy keys=[[#0{r1}, #1{c1}], [#3, #4]] // { arity: 5 } Get l14 // { arity: 5 } ArrangeBy keys=[[#0, #1]] // { arity: 2 } Get l5 // { arity: 2 } ArrangeBy keys=[[#0, #1]] // { arity: 3 } Union // { arity: 3 } Get l9 // { arity: 3 } Map (0) // { arity: 3 } Union // { arity: 2 } Negate // { arity: 2 } Project (#0, #1) // { arity: 2 } Get l9 // { arity: 3 } Get l6 // { arity: 2 } Constant // { arity: 4 } - (12, 20, 0, 0) Return // { arity: 4 } Get l15 // { arity: 4 } Source materialize.public.input Target cluster: quickstart EOF