-- Волт

-- 1) - проверенный рабочий вариант
-- у самой первой версии выгрузки вольта надо проставить дату в далёком прошлом. Для этого добавляем фиктивный клон самой первой версии со старой датой. Самую первую версию будем определять по самой ранней дате выгрузки
INSERT INTO TABLE core.vault
SELECT reflect("java.util.UUID", "randomUUID"), reflect("java.util.UUID", "randomUUID"), "ATMain", t.name, t.fullname, t.moneyorstock, t.vaultnumber,
t.isenable, t.vaultid, NULL AS isDeleted, "2000-01-01" AS start_date
FROM (
    SELECT v.name, v.fullname, v.moneyorstock, v.vaultnumber, v.isenable, v.vaultid, v.SOURCE, MIN(startDate) AS startDate
    FROM (
        SELECT v.name, v.fullname, v.moneyorstock, v.vaultnumber, v.isenable, v.vaultid, v.SOURCE, concat_ws("-", YEAR, MONTH, DAY) AS startDate
        FROM raw.atmain_vault v
    ) v
    GROUP BY v.name, v.fullname, v.moneyorstock, v.vaultnumber, v.isenable, v.vaultid, v.SOURCE
) t

-- 2) - вариант с косяками и багами (в процессе исправления)
-- Нужно учитывать:
-- при загрузке от разных дат, если волт не менялся, то новая запись в core не должна появиться, т.е. по сути - это могла быть такая же запись, с новым Version_ID и новым stardDate. И вот если такая ситуация, то не нужно делать новую запись. Это требование порождает UNION ALL
-- Для всех версий одного волта нужно проставлять один глобальный идентификатор Primary_ID.
INSERT OVERWRITE TABLE core.vault
SELECT IF(vv.primary_id IS NULL, reflect("java.util.UUID", "randomUUID"), vv.primary_id),
reflect("java.util.UUID", "randomUUID") AS version_id, "ATMain", f.name, f.fullname, f.moneyorstock, f.vaultnumber, f.isenable, f.sourcevault_id, f.isdeleted, f.startDate
FROM (
    SELECT DISTINCT f.name, f.fullname, f.moneyorstock, f.vaultnumber, f.isenable, f.sourcevault_id, f.isdeleted, f.row_hash, f.startDate
    FROM (
        SELECT t.name, t.fullname, t.moneyorstock, t.vaultnumber, t.isenable, t.sourcevault_id, t.isdeleted, t.row_hash,
        FIRST_VALUE(startDate) OVER (PARTITION BY t.name, t.fullname, t.moneyorstock, t.vaultnumber, t.isenable, t.sourcevault_id, t.isdeleted, t.row_hash ORDER BY t.tmp) AS startDate
        FROM (
            SELECT u.name, u.fullname, u.moneyorstock, u.vaultnumber, u.isenable, u.sourcevault_id, u.isdeleted, hash(concat_ws("_",
                IF (u.name IS NOT NULL AND u.name <> '', u.name, "null"),
                IF (u.fullname IS NOT NULL AND u.fullname <> '', u.fullname, "null"),
                IF (u.moneyorstock IS NOT NULL AND u.moneyorstock <> '', u.moneyorstock, "null"),
                IF (u.vaultnumber IS NOT NULL AND u.vaultnumber <> '', u.vaultnumber, "null"),
                CAST(u.isenable AS STRING),
                CAST(u.sourcevault_id AS STRING)
            )) AS row_hash, u.startDate, 1 AS tmp
            FROM core.vault u
            UNION ALL
            SELECT t.name, t.fullname, t.moneyorstock, t.vaultnumber, t.isenable, t.sourcevault_id, t.isdeleted, t.row_hash, t.startDate, 2 AS tmp
            FROM (
                SELECT *
                FROM(
                    SELECT name, fullname, moneyorstock, vaultnumber, isenable, vaultid AS sourcevault_id, NULL AS isdeleted, row_hash, concat_ws("-", YEAR, MONTH, DAY) AS startDate,
                    lag(row_hash, 1) OVER (ORDER BY vaultid, YEAR, MONTH, DAY) AS prev_row_hash
                    FROM (
                        SELECT u.*, hash(concat_ws("_",
                            IF (u.name IS NOT NULL AND u.name <> '', u.name, "null"),
                            IF (u.fullname IS NOT NULL AND u.fullname <> '', u.fullname, "null"),
                            IF (u.moneyorstock IS NOT NULL AND u.moneyorstock <> '', u.moneyorstock, "null"),
                            IF (u.vaultnumber IS NOT NULL AND u.vaultnumber <> '', u.vaultnumber, "null"),
                            CAST(u.isenable AS STRING),
                            CAST(u.vaultid AS STRING)
                        )) AS row_hash
                        FROM raw.atmain_vault u
                        ORDER BY u.vaultid, u.YEAR, u.MONTH, u.DAY
                    ) t
                )t
                WHERE (t.prev_row_hash IS NULL OR t.prev_row_hash <> t.row_hash)
            )t
        ) t
    ) f
)f
LEFT JOIN (
    SELECT primary_id, sourcevault_id
    FROM core.vault
    GROUP BY primary_id, sourcevault_id
    ORDER BY sourcevault_id
) vv ON f.sourcevault_id = vv.sourcevault_id;

-- ВАЖНО: version_id при таком переносе будет все время меняться. Но это не критично, т.к. ссылки на сущности определяются по primary_id

-- ВАЖНО: если, например, в core.vault не было vaultid = 100500, и в raw.atmain_vault есть сразу две версии с таким идентификатором, то primary_id у них прставится разный. Как это исправить?