Part One
预设了一些墙，如下文所示。模拟沙子下落的过程，计算最终有多少沙子处于静止状态
498,4 -> 498,6 -> 496,6
503,4 -> 502,4 -> 502,9 -> 494,9
计算墙的坐标
CREATE TABLE bricks
AS
WITH recursive seq AS (
SELECT 1 as id
UNION ALL
SELECT id + 1 as id
FROM seq
), origin AS (
SELECT start_x, start_y, end_x, end_y, abs(start_x - end_x) + abs(start_y - end_y) + 1 as steps
FROM (
SELECT _group, split_part(item, ',', 2) :: integer as start_x, split_part(item, ',', 1) :: integer as start_y,
split_part(lead(item) over (partition by _group), ',', 2) :: integer as end_x,
split_part(lead(item) over (partition by _group), ',', 1) :: integer as end_y
FROM (
SELECT _group, string_to_table(line, ' -> ') as item
FROM (
SELECT row_number() over () as _group, line
FROM lance_input
) t
) x
) t
WHERE end_x is not null
), origin_expand AS (
SELECT distinct
case when start_x < end_x then start_x + y.id - 1 when start_x > end_x then start_x - y.id + 1 else start_x end as pos_x,
case when start_y < end_y then start_y + y.id - 1 when start_y > end_y then start_y - y.id + 1 else start_y end as pos_y,
'rock' as type
FROM origin x,
(select id from seq limit (select max(steps) from origin)) y
WHERE y.id <= x.steps
)
SELECT * FROM origin_expand;
通过计算后，得出墙上每一块砖的具体坐标，如下所示
postgres=# select * From bricks;
pos_x | pos_y | type
-------+-------+------
4 | 503 | rock
9 | 494 | rock
6 | 496 | rock
9 | 501 | rock
9 | 496 | rock
4 | 502 | rock
8 | 502 | rock
9 | 498 | rock
6 | 498 | rock
6 | 502 | rock
9 | 497 | rock
9 | 495 | rock
5 | 498 | rock
5 | 502 | rock
4 | 498 | rock
9 | 502 | rock
6 | 497 | rock
9 | 500 | rock
7 | 502 | rock
9 | 499 | rock
(20 rows)
计算沙子的下落过程
当沙子下落时，按照down / down_left / down_right的优先级，从而得出下一步的位置。当三个位置均被占用时，则沙子无法继续下落，从而停止在当前位置。如果下落深度已经超过墙的最大深度（例子中是9），则代表沙子后续将持续下落，无法停止。
WITH recursive sand AS (
SELECT 0 as pos_x, 500 as pos_y, false as rest
UNION ALL
SELECT case when n1.pos_x is null then m.pos_x + 1 when n2.pos_x is null then m.pos_x + 1 when n3.pos_x is null then m.pos_x + 1 else m.pos_x end as pos_x,
case when n1.pos_y is null then m.pos_y when n2.pos_y is null then m.pos_y - 1 when n3.pos_y is null then m.pos_y + 1 else m.pos_y end as pos_y,
n1.pos_y is not null and n2.pos_y is not null and n3.pos_y is not null as rest
FROM sand m
LEFT JOIN bricks n1
ON m.pos_x + 1 = n1.pos_x
AND m.pos_y = n1.pos_y
LEFT JOIN bricks n2
ON m.pos_x + 1 = n2.pos_x
AND m.pos_y - 1 = n2.pos_y
LEFT JOIN bricks n3
ON m.pos_x + 1 = n3.pos_x
AND m.pos_y + 1 = n3.pos_y
WHERE not m.rest
AND m.pos_x < 9
)
比如，第一粒沙子，其下落过程可以通过计算得出其最终停止的位置是(8, 500)
pos_x | pos_y | rest
-------+-------+------
0 | 500 | f
1 | 500 | f
2 | 500 | f
3 | 500 | f
4 | 500 | f
5 | 500 | f
6 | 500 | f
7 | 500 | f
8 | 500 | f
8 | 500 | t
(10 rows)
最终SQL
通过函数，持续迭代这个过程，直到没有静止的沙子为止
CREATE OR REPLACE FUNCTION move_sand()
RETURNS VOID AS $$
DECLARE
insert_cnt INTEGER;
BEGIN
WHILE TRUE LOOP
--insert sand
WITH recursive sand AS (
SELECT 0 as pos_x, 500 as pos_y, false as rest
UNION ALL
SELECT case when n1.pos_x is null then m.pos_x + 1 when n2.pos_x is null then m.pos_x + 1 when n3.pos_x is null then m.pos_x + 1 else m.pos_x end as pos_x,
case when n1.pos_y is null then m.pos_y when n2.pos_y is null then m.pos_y - 1 when n3.pos_y is null then m.pos_y + 1 else m.pos_y end as pos_y,
n1.pos_y is not null and n2.pos_y is not null and n3.pos_y is not null as rest
FROM sand m
LEFT JOIN bricks n1
ON m.pos_x + 1 = n1.pos_x
AND m.pos_y = n1.pos_y
LEFT JOIN bricks n2
ON m.pos_x + 1 = n2.pos_x
AND m.pos_y - 1 = n2.pos_y
LEFT JOIN bricks n3
ON m.pos_x + 1 = n3.pos_x
AND m.pos_y + 1 = n3.pos_y
WHERE not m.rest
AND m.pos_x < 168
), insert_sand AS (
INSERT INTO bricks
SELECT pos_x, pos_y, 'sand' as type
FROM sand
WHERE rest
RETURNING 1
)
SELECT COUNT(*) INTO insert_cnt FROM insert_sand;
IF insert_cnt < 1 THEN
EXIT;
END IF;
END LOOP;
END;
$$ LANGUAGE plpgsql;
最终，通过过滤type=sand的记录，来求出静止的沙子总数
postgres=# select move_sand();
move_sand
-----------
(1 row)
postgres=# select count(*) from bricks where type = 'sand';
count
-------
1330
(1 row)
Part Two
第二关，增加了地板，最终沙子总会静止，直到入口(0, 500)也被占据。
assume the floor is an infinite horizontal line with a y coordinate equal to two plus the highest y coordinate
Pyramid
增加了地板之后，总体就会呈现等边三角形的样式。
因此判断是否静止的条件也需要随之调整，主要就是：
- x为地板上一层时，肯定为静止状态
- x为地板上一层时，则不再继续往下寻找
SELECT case when n1.pos_x is null and m.pos_x < 169 then m.pos_x + 1
when n2.pos_x is null and m.pos_x < 169 then m.pos_x + 1
when n3.pos_x is null and m.pos_x < 169 then m.pos_x + 1
else m.pos_x
end as pos_x,
case when n1.pos_y is null and m.pos_x < 169 then m.pos_y
when n2.pos_y is null and m.pos_x < 169 then m.pos_y - 1
when n3.pos_y is null and m.pos_x < 169 then m.pos_y + 1
else m.pos_y
end as pos_y,
m.pos_x = 169 OR (n1.pos_y is not null and n2.pos_y is not null and n3.pos_y is not null) as rest
但是，当运行实际数据的时候，每次仅插入最后一条记录，而且越往后速度越慢，导致长时间运行没有结果。
一次插入多条
假如目前的沙子形状为下图
.....o.....
....ooo....
...ooooo...
再流入三粒沙子后，应该变为如下
....oo.....
...oooo....
..oooooo...
再流入三粒沙子后，应该变为如下
....ooo....
...ooooo...
..ooooooo..
可见，左边新增的边，以及右边新增的边，其实可以一次性插入，而无需执行三次逐个插入。因此在计算过程中，通过flag字段来记录是否为斜边上连续的记录，如果是连续的记录，则最终可以一次性全部插入。
direction代表下落的方向，D为垂直下落，L为左下，R为右下。
case when (direction = 'L' AND pre_y > pos_y) OR (direction = 'R' AND pre_y < pos_y) OR rest then flag else flag + 1 end as flag,
case when pre_y = pos_y then 'D' when pre_y > pos_y then 'L' else 'R' end as direction
最终插入的时候，可以根据flag插入多行
INSERT INTO bricks
SELECT distinct pos_x, pos_y, 'sand' as type
FROM sand
WHERE flag in (select flag from sand where rest)
RETURNING pos_x
修正算法
在计算示例数据的时候，缺失了两个点
其实，就是因为在计算的时候，直接插入了整条边，导致没有考虑还有右侧的潜在位置。
因此，需要增加一个判断条件，只有only_pos为true的时候，才可以继续插入。
case when n2.pos_x is null and n3.pos_x is not null then true
when n2.pos_x is not null and n3.pos_x is null then true
else false
end as only_pos
case when (direction = 'L' AND pre_y > pos_y AND only_pos) OR (direction = 'R' AND pre_y < pos_y AND only_pos) OR rest then flag else flag + 1 end as flag
完整SQL
CREATE OR REPLACE FUNCTION move_sand()
RETURNS VOID AS $$
DECLARE
sand_x INTEGER;
BEGIN
WHILE TRUE LOOP
--insert sand
WITH recursive sand AS (
SELECT 0 as pos_x, 500 as pos_y, false as rest, 1 as flag, 'D' as direction
UNION ALL
SELECT pos_x, pos_y, rest,
case when (direction = 'L' AND pre_y > pos_y AND only_pos) OR (direction = 'R' AND pre_y < pos_y AND only_pos) OR rest then flag else flag + 1 end as flag,
case when pre_y = pos_y then 'D' when pre_y > pos_y then 'L' else 'R' end as direction
FROM (
SELECT m.flag, m.pos_x as pre_x, m.pos_y as pre_y, m.direction,
case when n1.pos_x is null and m.pos_x < 169 then m.pos_x + 1
when n2.pos_x is null and m.pos_x < 169 then m.pos_x + 1
when n3.pos_x is null and m.pos_x < 169 then m.pos_x + 1
else m.pos_x
end as pos_x,
case when n1.pos_y is null and m.pos_x < 169 then m.pos_y
when n2.pos_y is null and m.pos_x < 169 then m.pos_y - 1
when n3.pos_y is null and m.pos_x < 169 then m.pos_y + 1
else m.pos_y
end as pos_y,
m.pos_x = 169 OR (n1.pos_y is not null and n2.pos_y is not null and n3.pos_y is not null) as rest,
case when n2.pos_x is null and n3.pos_x is not null then true
when n2.pos_x is not null and n3.pos_x is null then true
else false
end as only_pos
FROM sand m
LEFT JOIN bricks n1
ON m.pos_x + 1 = n1.pos_x
AND m.pos_y = n1.pos_y
LEFT JOIN bricks n2
ON m.pos_x + 1 = n2.pos_x
AND m.pos_y - 1 = n2.pos_y
LEFT JOIN bricks n3
ON m.pos_x + 1 = n3.pos_x
AND m.pos_y + 1 = n3.pos_y
WHERE not m.rest
) t
), insert_sand AS (
INSERT INTO bricks
SELECT distinct pos_x, pos_y, 'sand' as type
FROM sand
WHERE flag in (select flag from sand where rest)
RETURNING pos_x
)
SELECT pos_x INTO sand_x FROM sand where rest;
raise notice '%', sand_x;
IF sand_x = 0 THEN
EXIT;
END IF;
END LOOP;
END;
$$ LANGUAGE plpgsql;