如何在BigQuery SQL中实现有序列间隙不超阈值的行筛选

给定如下数据集：

WITH xs AS (
    SELECT 'A' name, 21 value
    UNION ALL
    SELECT 'B', 24
    UNION ALL
    SELECT 'C', 28
    UNION ALL
    SELECT 'D', 47
    UNION ALL
    SELECT 'E', 48
    UNION ALL
    SELECT 'F', 49
    UNION ALL
    SELECT 'G', 50
    UNION ALL
    SELECT 'H', 58
)

需要筛选出满足以下条件的行：确保排序后的value列不会出现超过10的间隙。具体逻辑已用Python实现：

from dataclasses import dataclass

max_gap = 10

@dataclass
class Row:
    name: str
    value: int

xs = [Row('A', 21), Row('B', 24), Row('C', 28), Row('D', 47), Row('E', 48), Row('F', 49), Row('G', 50), Row('H', 58)]

ys = [xs[0]]
for idx in range(1, len(xs) - 1):
    if xs[idx + 1].value > ys[-1].value + max_gap:
        ys.append(xs[idx])
ys.append(xs[-1])

assert ys == [Row('A', 21), Row('C', 28), Row('D', 47), Row('G', 50), Row('H', 58)]

但无法将该逻辑转换为BigQuery SQL，核心难点在于：每行的筛选判断不仅依赖相邻行（常规可用WINDOW结合LEAD/LAG处理），还依赖输出结果集中的最后一个值。不确定递归CTE、过程语言（LOOP/FOR）还是常规SQL逻辑能实现，求具体的BigQuery实现方案。

可以用递归CTE来实现这个逻辑，它能追踪上一个被保留的行的value值，逐行判断是否需要保留当前行。具体代码如下：

WITH xs AS (
    SELECT 'A' name, 21 value
    UNION ALL
    SELECT 'B', 24
    UNION ALL
    SELECT 'C', 28
    UNION ALL
    SELECT 'D', 47
    UNION ALL
    SELECT 'E', 48
    UNION ALL
    SELECT 'F', 49
    UNION ALL
    SELECT 'G', 50
    UNION ALL
    SELECT 'H', 58
),
-- 给每行添加序号，方便递归遍历
ordered_xs AS (
    SELECT 
        name, 
        value,
        ROW_NUMBER() OVER(ORDER BY value) AS rn
    FROM xs
),
-- 递归CTE：初始化保留第一行
recursive_selection AS (
    SELECT 
        name, 
        value, 
        rn,
        value AS last_kept_value  -- 记录上一个被保留的value
    FROM ordered_xs
    WHERE rn = 1

    UNION ALL

    SELECT 
        curr.name, 
        curr.value, 
        curr.rn,
        -- 如果当前行需要保留，更新last_kept_value为当前value，否则沿用之前的
        CASE 
            WHEN next.value > prev.last_kept_value + 10 THEN curr.value 
            ELSE prev.last_kept_value 
        END AS last_kept_value
    FROM recursive_selection prev
    JOIN ordered_xs curr 
        ON curr.rn = prev.rn + 1
    LEFT JOIN ordered_xs next 
        ON next.rn = curr.rn + 1
    -- 只处理到倒数第二行，最后一行单独添加
    WHERE curr.rn < (SELECT MAX(rn) FROM ordered_xs)
),
-- 收集递归结果中的保留行
kept_rows AS (
    SELECT name, value
    FROM recursive_selection
    WHERE value = last_kept_value

    UNION ALL

    -- 添加最后一行
    SELECT name, value
    FROM ordered_xs
    WHERE rn = (SELECT MAX(rn) FROM ordered_xs)
)
SELECT name, value
FROM kept_rows
ORDER BY value;

逻辑说明 #

1. ordered_xs：先给数据集按value排序并添加行号，确保递归能按顺序遍历。
2. recursive_selection：
   • 初始步骤：保留第一行，记录其value为last_kept_value。
   • 递归步骤：每次取当前行和下一行，判断下一行的value是否超过上一个保留值+10。如果是，就保留当前行，并更新last_kept_value为当前行的value；否则不保留当前行，沿用之前的last_kept_value。
3. kept_rows：从递归结果中筛选出所有被保留的行（即value等于last_kept_value的行），再加上最后一行，最终得到符合要求的结果集。

运行上述代码后，输出结果与Python实现的断言一致：

内容的提问来源于stack exchange，提问作者Tobias Hermann