Как построить этот запрос BigQuery?

Question

У меня есть набор данных, который выглядит примерно так:

WITH alignments AS (
    SELECT 1 AS id, 'Chaotic' AS name, '1999-01-01 00:00:00' AS ingestion
    UNION ALL
    SELECT 2, 'Good', '1970-01-01 00:00:00'
    UNION ALL
    SELECT 3,'Evil','1970-01-01 00:00:00'
    UNION ALL
    SELECT 4,'Annoying','2003-01-01 00:00:00'
    UNION ALL
    SELECT 5,'Whiny','1970-01-01 00:00:00'
),


WITH species AS (
    SELECT 1 AS id, 'Human' AS name, '1970-01-01 00:00:00' AS ingestion
    UNION ALL
    SELECT 2, 'Gun Gun', '2000-01-01 00:00:00
),

WITH characters AS (
    SELECT 1 AS id, 'Jar Jar Binks' AS name, 2 AS species, 1 AS alignment, '2000-01-01 01:00:00' AS ingestion
    UNION ALL
    SELECT 1, 'Jar Jar Binks', 2, 1, '2001-01-02 02:00:00'
    UNION ALL
    SELECT 2, 'Qui Gon Jinn', 1, 2, '2000-01-01 01:00:00'
    UNION ALL
    SELECT 3, 'Obi Wan Kenobi', 1, 2, '1970-01-01 01:30:00'
    UNION ALL
    SELECT 4, 'Darth Vader', 1, 3, '1970-01-01 05:00:00'
    UNION ALL
    SELECT 5, 'Luke Skywalker', 1, 5, '1970-01-01 04:00:00'
    UNION ALL
    SELECT 6, 'Leia Organa', 1, 2, '1970-01-01 04:00:01'
    UNION ALL
    SELECT 7, 'Anakin Skywalker', 1, 2, '2001-01-01 00:00:06'
    UNION ALL
    SELECT 7, 'Anakin Skywalker', 1, 5, '2003-01-01 00:00:01
    UNION ALL
    SELECT 8, 'Rey', 1, 2, '2016-01-01 01:05:45'
),

WITH movies AS (
    SELECT 1 AS id, 'A New Hope' AS title, '.75' AS rotten_tomatoes_score, '1970-01-01 00:00:00' AS ingestion
    UNION ALL
    SELECT 2, 'The Empire Strikes Back', '.89', '1974-01-01 01:20:00'
    UNION ALL
    SELECT 3, 'Return of the Jedi', '.69', '1981-01-03 06:07:59'
    UNION ALL
    SELECT 3, 'Return of the Jedi', '.70', '1982-03-04 08:15:45'
    UNION ALL
    SELECT 4, 'The Phantom Menace', '.10', '2001-01-01 00:00:00
    UNION ALL
    SELECT 1, 'A New Hope', '.85', '1970-01-01 00:00:00'
    UNION ALL
    SELECT 5, 'The Clone Wars', '.35', '2003-01-01 00:00:00'
    UNION ALL
    SELECT 6, 'Revenge of the Sith', '.65', '2005-01-01 00:00:00'
    UNION ALL
    SELECT 7, 'The Force Awakens', '.78', '2016-01-01 01:04:00'
    UNION ALL
    SELECT 8, 'Return of the Jedi', '.10', '2018-01-01 00:00:00'
),

WITH characters_movies AS (

    SELECT 1 AS character_id, 4 AS movie_id, '2000-01-01 01:00:00' AS ingestion:
    UNION ALL
    SELECT 2, 7, 4, '2000-01-01 01:00:00'
    ....
)

Учитывая следующие требования:

• Мне нужно создать снимок фильмов, которые были изменены в пределах окно загрузки.
• Каждая запись снимка должна содержать mov ie и массив символов с указанием их выравнивания и вида.
• Если изменяется запись в символах, фильмах, character_movies, тогда любой прикрепленный mov ie должен быть включен в снимок.
• Для любого mov ie, включенного в окно загрузки, необходимо получить последние связанные записи, которые существовали на момент MOV ie был проглочен. Таким образом, теоретически создание моментального снимка будет идемпотентом.

Я знаком с тем, как выполнить sh # 2. Пример:

SELECT
    id AS movie_id,
    title,
    rotten_tomatoes_score,
    ingestion
    ARRAY_AGG(STRUCT(
        c.id,
        c.name,
        a.name,
        s.name,
        c.ingestion
    )) AS characters
    FROM movies m
    LEFT JOIN characters_movies cm ON m.id = cm.movie_id
    LEFT JOIN characters c ON cm.character_id = c.id
    LEFT JOIN alignments a ON c.alignment = a.id
    LEFT JOIN species s ON c.species = s.id
    GROUP BY m.id, m.title, m.rotten_tomatoes_score

Для # 3 я почти уверен, что могу использовать выражение запроса WITH для генерации списка всех идентификаторов mov ie, включаемых в снимок. Пример:

WITH ids AS (
    SELECT id AS movie_id FROM movies WHERE ingestion BETWEEN ts1 AND ts2
    UNION DISTINCT
    SELECT movie_id
    FROM character_movies cm
    WHERE ingestion BETWEEN ts1 AND ts2
    UNION DISTINCT
    SELECT movie_id
    FROM character_movies cm
    JOIN characters c ON cm.character_id = c.character_id
    WHERE cm.ingestion < ts2
    AND c.ingestion BETWEEN ts1 AND ts2
)

Как правило, я нашел это решение для поиска самой последней версии записи в данном окне времени. Пример:

SELECT *
FROM (
    SELECT *, ROW_NNUMBER() OVER (PARTITION BY id ORDER BY ingestion DESC) AS rn
    FROM movies m
    JOIN ids ON m.id = ids.movie_id
) t
WHERE t.rn = 1

Но это не решает проблему ограничения меня только изменениями, которые произошли бы в окне приема, просто захватывает самую последнюю запись для каждого mov ie, который будет включен в окне, но не позволяет мне открыть версию каждой записи, которая будет иметь значение только go до конца указанного окна приема, но позволяет включать данные старше, чем начало окна, если версия в окно не существует.