CALL apoc.periodic.iterate(
    '
        WITH "
            SELECT
                sc.rank,
                sc.timestp,
                TRIM(sc.country) AS country,
                sc.chart_type,
                sc.duration,
                UPPER(st.isrc) AS isrc,
                cmt.name AS sound_recording_name,
                cmt.image_url AS sound_recording_image_url,
                st.spotify_track_id,
                st.track_name,
                st.artwork_url AS track_image_url,
                st.spotify_album_id,
                sp.name AS product_name,
                sp.artwork_url AS product_image_url,
                sp.upc,
                sp.release_date,
                sp.label,
                cma.id AS cm_artist_id
            FROM chartmetric.raw_data.spotify_chart sc
            JOIN chartmetric.raw_data.spotify st ON st.id = sc.spotify
            JOIN chartmetric.raw_data.cm_track cmt ON UPPER(cmt.isrc) = UPPER(st.isrc)
            LEFT JOIN chartmetric.raw_data.spotify_album sp ON sp.spotify_album_id = st.spotify_album_id
            JOIN chartmetric.raw_data.l_spotify_artist lsta ON lsta.spotify = st.id
            JOIN chartmetric.raw_data.spotify_artist sa ON sa.id = lsta.spotify_artist
            JOIN chartmetric.raw_data.cm_artist cma ON cma.id = sa.cm_artist
            WHERE sc.timestp >= \'2020-01-01\'
            AND st.isrc IS NOT NULL
        " AS sql
        CALL apoc.load.jdbc("snowflake_orchard", sql) YIELD row RETURN row
    ',
    '
        MERGE (sr:SoundRecording:Chartmetric {isrc: row.ISRC})
            ON CREATE SET
                sr.id = apoc.create.uuid(),
                sr.createdAt = datetime()
            SET
                sr.name = row.SOUND_RECORDING_NAME,
                sr.imageUrl = row.SOUND_RECORDING_IMAGE_URL,
                sr.modifiedAt = datetime()
        MERGE (t:Track:ChartmetricSpotify {spotifyId: row.SPOTIFY_TRACK_ID})
            ON CREATE SET
                t.id = apoc.create.uuid(),
                t.createdAt = datetime()
            SET
                t.name = row.TRACK_NAME,
                t.imageUrl = row.TRACK_IMAGE_URL,
                t.modifiedAt = datetime()
        MERGE (pr:Product:ChartmetricSpotify {spotifyId: row.SPOTIFY_ALBUM_ID})
            ON CREATE SET
                pr.id = apoc.create.uuid(),
                pr.createdAt = datetime()
            SET
                pr.name = row.PRODUCT_NAME,
                pr.imageUrl = row.PRODUCT_IMAGE_URL,
                pr.upc = row.UPC,
                pr.releaseDate = date(row.RELEASE_DATE),
                pr.modifiedAt = datetime()
        FOREACH (n IN (CASE WHEN row.LABEL IS NOT NULL THEN [1] else [] END) |
            MERGE (l:Label:ChartmetricSpotify {name: row.LABEL})
                ON CREATE SET
                    l.id = apoc.create.uuid(),
                    l.createdAt = datetime()
            MERGE (l)-[:OWNS_PRODUCT]->(pr)
        )
        MERGE (c:Chart:Chartmetric {platform: \'spotify\', target: \'track\', country: row.COUNTRY, type: row.CHART_TYPE, frequency: row.DURATION})
            ON CREATE SET
                c.id = apoc.create.uuid(),
                c.createdAt = datetime()
        WITH row, sr, t, pr, c
        MATCH(p:Participant:Chartmetric {chartmetricId: row.CM_ARTIST_ID})
        MERGE (p)-[:PARTICIPATED_IN]->(sr)
        MERGE (t)-[:IS_PLACEMENT_OF]->(sr)
        MERGE (pr)-[:INCLUDES]->(t)
        MERGE (sr)-[ro:RANKED_ON {timestamp: date(row.TIMESTP)}]->(c)
            ON CREATE SET
                ro.createdAt = datetime()
            SET
                ro.position = row.RANK,
                ro.modifiedAt = datetime()
    ',
    {batchSize:1000, iterateList:true, parallel:false}
);