activityRelations_snapshot_merger_copy.pipe

DESCRIPTION >
    Merges realtime snapshots (coming from MV) and historical snapshot (coming from datasource) to create another snapshot

NODE realtime_snapshot
SQL >
    WITH (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) as maxSnapshotId
    SELECT activityRelations_enrich_snapshot_MV_ds.*
    FROM activityRelations_enrich_snapshot_MV_ds FINAL
    where activityRelations_enrich_snapshot_MV_ds.snapshotId = maxSnapshotId + INTERVAL 1 day

NODE historical_snapshot
SQL >
    WITH
        (
            Select max(snapshotId)::DateTime from activityRelations_enriched_deduplicated_ds
        ) as maxSnapshotID
    SELECT h.* REPLACE (maxSnapshotID::DateTime + interval 1 day AS snapshotId)
    FROM activityRelations_enriched_deduplicated_ds h
    LEFT ANTI JOIN realtime_snapshot r
        ON  h.segmentId  = r.segmentId
        AND h.timestamp  = r.timestamp
        AND h.type       = r.type
        AND h.platform   = r.platform
        AND h.channel    = r.channel
        AND h.sourceId   = r.sourceId
    WHERE h.snapshotId = maxSnapshotID
    SETTINGS join_algorithm = 'grace_hash'

NODE merged_snapshots
SQL >
    select *
    from realtime_snapshot
    union all
    select *
    from historical_snapshot

TYPE COPY
TARGET_DATASOURCE activityRelations_enriched_deduplicated_ds
COPY_MODE append
COPY_SCHEDULE 0 1 * * *