CALL apoc.periodic.iterate(
    '
        WITH "
            SELECT
                amc.rank,
                amc.timestp,
                TRIM(amc.country) AS country,
                UPPER(it.isrc) AS isrc,
                cmt.name AS sound_recording_name,
                cmt.image_url AS sound_recording_image_url,
                it.itunes_track_id,
                it.track_name,
                it.artwork_url AS track_image_url,
                it.itunes_album_id,
                itp.name AS product_name,
                itp.artwork_url AS product_image_url,
                itp.upc,
                itp.release_date,
                IFNULL(itp.label, itp.copyright) AS label,
                cma.id AS cm_artist_id
            FROM chartmetric.raw_data.apple_music_daily_top_chart amc
            JOIN chartmetric.raw_data.itunes it ON it.id = amc.itunes
            JOIN chartmetric.raw_data.cm_track cmt ON cmt.isrc = it.isrc
            LEFT JOIN chartmetric.raw_data.itunes_album itp ON itp.itunes_album_id = it.itunes_album_id
            JOIN chartmetric.raw_data.l_itunes_artist lita ON lita.itunes = it.id
            JOIN chartmetric.raw_data.itunes_artist ita ON ita.id = lita.itunes_artist
            JOIN chartmetric.raw_data.cm_artist cma ON cma.id = ita.cm_artist
            WHERE amc.timestp >= \'2020-01-01\'
            AND it.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:ChartmetricAppleMusic {appleMusicId: row.ITUNES_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:ChartmetricAppleMusic {appleMusicId: row.ITUNES_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:ChartmetricAppleMusic {name: row.LABEL})
                ON CREATE SET
                    l.id = apoc.create.uuid(),
                    l.createdAt = datetime()
            MERGE (l)-[:OWNS_PRODUCT]->(pr)
        )
        MERGE (c:Chart:Chartmetric {platform: \'appleMusic\', target: \'track\', country: row.COUNTRY, type: \'default\', frequency: \'daily\'})
            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}
);