setScience("Spotify_Accounting_ETL", subl=FALSE)
setDBall(cluster=4)


frmt.create_with_label <- {
"   CREATE TABLE bi.%s
   (
     owner VARCHAR (11),
     isdistributor VARCHAR (2),
     labelid INT4,
     genreid INT2,
     upc NUMERIC (22, 6),
     customerid VARCHAR(35),
     product VARCHAR(2),
     country VARCHAR(3)
   )

   DISTSTYLE EVEN

   SORTKEY
   (
      upc
   )
"}



tbl_spot <- "staging_raw_spotify"

maxDates <- seq.Date(as.Date("2013-07-01"), by="3 months", length.out=12) - 1
maxDates <- maxDates[maxDates < today()]

nms <- format(maxDates, "upto_%Y%m%d")
setattr(maxDates, "names", nms)
setattr(nms,      "names", nms)


Qrys.full <- lapply(nms, function(nm) {
  setQry(sprintf(
    "SELECT L.owner, L.isdistributor, TT.*
    FROM (
      SELECT R.labelid, R.genreid, S.*
      FROM ( %s 
          ) S
      LEFT JOIN production.dim_release R
      ON S.upc = R.releaseid
    ) TT
    LEFT JOIN production.dim_label L
    ON TT.labelid = L.labelid
    ", 
      makeQry(tbl=tbl_spot, schema="production", colsToPull=c("upc", "customerid"
      , product="CASE WHEN product IN ('O', 'F') THEN 'F' ELSE PRODUCT END", "country"), colsToAgg=NULL, distinct=TRUE, limit=NULL, maxDate=maxDates[[nm]], dateCol="download_date", minDate="2013-04-01")
  ))
})


setDBall(cluster=4)
for (nm in names(maxDates)) {
  bucket <- s3_p("spotify", "FO_users_per_release_country", nm)
  unloadQry(qry=Qrys.full[[nm]], bucket=bucket)
}


setDBall(cluster=9)
for (nm in names(maxDates)) {
  tbl <- paste0("labels_and_customers_", nm)
  cat("tbl = ", tbl, "\n")
  try(runQry(sprintf("DROP TABLE bi.%s", tbl)))
  runQry(sprintf(frmt.create_with_label, tbl))

  bucket <- s3_p("spotify", "FO_users_per_release_country", nm)
  loadFromBucket(bucket=bucket, tbl=tbl, schema="bi")
}



ll_DT.customer_counts_by_label <- emptylist(maxDates)
for (nm in names(maxDates)[1]) {
  tbl <- paste0("labels_and_customers_", nm)
  Qry <- makeQry(tbl=tbl, schema="bi", colsToPull=c("owner", "labelid", "product", "country"), colsToAgg=setNames(paste0("customer_count_", nm), obj="DISTINCT customerid"), aggFunc="count", limit=NULL, key="colsToPull")
  ll_DT.customer_counts_by_label[[nm]] <- runQry(Qry, verbose=TRUE)
}


DT.customer_counts_by_label <- Reduce(f=function(L, R) data.table:::merge.data.table(L, R, all=TRUE), x=rev(ll_DT.customer_counts_by_label))


jesusForData("DT.customer_counts_by_label")