setScience("misc")
setGitBranchToSystem(); .g()

wh <- "Science"
dbname <- "prod"
setSnowflake(wh=wh, dbname=dbname)

schemas <- sfShowSchemas(dbname=dbname, justnames=TRUE) %>% tolower
DT.tbls <- sfShowTables(dbname=dbname, cleanBytes=FALSE)

## Clean up
setnames(DT.tbls, "schema_name", "schema")
DT.tbls[, name := tolower(name)]
DT.tbls[, schema := tolower(schema)]
setkey(DT.tbls, schema, name)
DT.tbls[.("youtube_analytics"), comment := NA_character_]
DT.tbls[.("spotifymgmt", "currency_neutral_factors"), comment := NA_character_]

## Which schemas
schemas_not_checking <- c("art_relations")

DT.tbls[.("art_relations")]
DT.tbls[rows > 100000]

DT.tbls[order(bytes, decreasing=TRUE)][1:30]

ll.tbls <- list()
## Production
ll.tbls[["production"]] <- c("staging_raw_apple_music_streams", "staging_raw_spotify_v2", "fact_analytics", "fact_analytics_error", "fact_sales", "staging_raw_apple_music")
## Schema
ll.tbls[["dim"]] <- c("dim_track", "dim_isrc", "dim_release", "dim_catalog", "dim_artist", "dim_imprint", "dim_subaccount", "dim_label")
## youtube_analytics
ll.tbls[["youtube_analytics"]] <- c("assetcombined_master", "assetdeviceos_master", "combined_master", "assetbasic_master", "deviceos_master", "assetplaybacklocation_master", "basic_master", "playbacklocation_master", "assetprovince_master", "assetdemographics_master")
## bi
ll.tbls[["bi"]] <- c("analytics_track_level", "analytics_release_level", "accounting")

## Other tables, FYI
DT.tbls[!.(names(ll.tbls))][order(bytes, decreasing=TRUE)][1:30]

if (!exists("ll.DT_rows")) {ll.DT_rows <- list()}

for (sch in names(ll.tbls)) {  
  ll.DT_rows[[sch]] <- setNames(nm=ll.tbls[[sch]], obj=lapply(ll.tbls[[sch]], function(tbl) sfQry(makeQry(tbl=tbl, schema=ifelse(sch=="dim", "production", sch), colsToPull=getDateColForTBL(schema=ifelse(sch=="dim", "production", sch), tbl=tbl), colsToAgg=c(rowcount="*"), aggFunc="count", limit=NULL, orderby=1))))
}

ll.DT_rows.bak <- copy(ll.DT_rows)

for (sch in names(ll.DT_rows)) {
  for (tbl in names(ll.DT_rows[[sch]])) {
    DT <- ll.DT_rows[[c(sch, tbl)]]
    dateCol <- names(DT)[[1]]
    setkeyIfNot(DT, dateCol, verbose=FALSE)
    DT[order(get(dateCol)), cum_rowcount := cumsum(as.numeric(rowcount))]
    if (grepl("periodid", dateCol)) {
      nms_old <- copy(names(DT))
      addDateCols.periodid_(DT, showWarnings=FALSE, dropPeriodCols=TRUE)
      dateCol <- setdiff(copy(names(DT)), nms_old)
      setkeyIfNot(DT, dateCol, organize=TRUE)
    }
    print(key(DT))
    DT[, schema := sch]
    DT[, tbl := tbl]
    # ll.DT_rows[[c(sch, tbl)]][order(ll.DT_rows[[c(sch, tbl)]][[1]]), cum_rowcount := cumsum(as.numeric(rowcount))]
  }
}

stack_and_agg <- function(ll.DT, dateCol) {
   ret <- {ll.DT %>% rbindlist(use.names=FALSE) %>% aggregateDT(by=c(date=dateCol, "schema"), exclude=c("tbl"), showWarnings.info=FALSE)}
   setnames(ret, names(ret)[[1]], "date")
   setkeyIfNot(ret, "date")
   ret[, cum_rowcount := cumsumn(rowcount)]
   when_doubled(ret, dateCol="date", countCol="cum_rowcount")
   when_halfed(ret, dateCol="date", countCol="cum_rowcount")
   return(ret)
}

DT.youtube <- ll.DT_rows[["youtube_analytics"]] %>% stack_and_agg(dateCol="date")
DT.applemusic <- ll.DT_rows[["production"]][c("staging_raw_apple_music", "staging_raw_apple_music_streams")] %>% stack_and_agg(dateCol="download_date")
DT.spotify <- ll.DT_rows[["production"]][c("staging_raw_spotify_v2")] %>% stack_and_agg(dateCol="tmstamp::date")
DT.analytics <- ll.DT_rows[["production"]][c("fact_analytics", "fact_analytics_error")] %>% stack_and_agg(dateCol="download_activity_date")
DT.accounting <- ll.DT_rows[["production"]][c("fact_sales")] %>% stack_and_agg(dateCol="accountingdate")


DT.youtube[, tbl := "youtube"]
DT.applemusic[, tbl := "applemusic"]
DT.spotify[, tbl := "spotify"]
DT.analytics[, tbl := "analytics"]
DT.accounting[, tbl := "accounting"]

DT.tables <- 
rbindlist(list(
  DT.youtube,
  DT.applemusic,
  DT.spotify,
  DT.analytics,
  DT.accounting
))

setnames(DT.tables, names(DT.tables)[[1]], "date")
setkeyIfNot(DT.tables, date, schema, tbl)
jesusForData(DT.tables)

## GRAPH
P.Total_Rows_Of_Data_Since_2012_by_table <- ggLinegraph(DT.tables[date >= "2012-01-01"], x="date", y="cum_rowcount", color="tbl", legend_title_on=FALSE, dotsize.scale=1.2, smooth=FALSE, dots=FALSE, xlab=NULL, ylab="Billions of Rows of Data", yscale="billions", title="\nTotal Rows of Data since 2012 by Table Source")
P.New_Rows_Of_Data_Since_2012_by_table <- ggLinegraph(DT.tables[date >= "2012-01-01"], x="date", y="rowcount", color="tbl", legend_title_on=FALSE, dotsize.scale=1.2, smooth=TRUE, dots=FALSE, xlab=NULL, ylab="Billions of Rows of Data", yscale="billions", title="\nNew Rows of Data since 2012 by Table Source")
## Save the graph
printToPDF(list(P.Total_Rows_Of_Data_Since_2012_by_table, P.New_Rows_Of_Data_Since_2012_by_table), "Data Growth 2016")

## Unused
DT.combined <- aggregateDT(DT.tables, by="date", exclude=c("schema", "tbl"), showWarnings.info=FALSE)
DT.combined_monthly <- aggregateMonthly(DT.combined)

P.New_Rows_Of_Data_Since <- 
  ggLinegraph(DT.combined_monthly[date >= "2014-01-01"], x="date", y="rowcount", legend_title_on=FALSE, dotsize.scale=1.2, smooth=FALSE, dots=TRUE, ylab="Billions of Rows of Data", yscale="billions", title="Rows of Data by Table Source")