# ~/git/orch/src/Discover_Weekly/supportFns/create_spotify_stream_counts_table supportFns.r create_spotify_stream_counts_table <- function( colsToPull , colsWithaggFunc = c(weekof = "friday_weekof(activity_date::date)") ## , date_last_streamed = "friday_weekof(max(last_tmstamp::date))" , dateCol.nm = names(colsWithaggFunc)[[1]] , dateCol_in = "tmstamp" , colsToAgg = c(streams="*") , tbl_out = "..auto.." , aggFunc = "sum" , tbl_in = "sos_from_raw_view" , schema = "spot_stream_counts" , schema_in = schema , maxDate = "..auto.." , confirm_streams = TRUE , uid_columns=c("anonymized_uid", "user_id", "uid") , minDate = "friday_weekof('2015-01-01'::date)" , having_clause = if (length(minDate) == 1) sprintf("tmstamp::date > %s", ifelse(grepl("^\\d", minDate), sprintf("'%s'", minDate), minDate)) , streams_expected = makeQry(tbl=tbl_in, schema=schema_in, dateCol=dateCol_in, maxDate=maxDate, colsToAgg="*", aggFunc="count", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist() , where__for_debugging=NULL , verbose.qry=FALSE ) { if (aggFunc != "sum") warning("Aggregation may be incorrect when using aggFunc other than 'sum'") ## We actually do NOT want uid_columns in here; We allow for them so that we can use the same for-loop if (any(uid_columns %in% colsToPull)) colsToPull %<>% setdiff(uid_columns) if (!is.null(where__for_debugging)) message("REMINDER: The WHERE clause should only be used for debugging. It will apply to the inner table") if (tbl_out == "..auto..") tbl_out <- create_counts_table_name_from_colsToPull(colsToPull=colsToPull, uid_columns=uid_columns) if (grepl("_test$", tbl_in, ignore.case=TRUE) && !grepl("_test", tbl_out, ignore.case=TRUE)) { verboseMsg(TRUE, "Pasting '_test' to name of tbl_out") tbl_out %<>% paste0("_TEST") } ## ------ COUNTS QUERY --------------------------------- ## This will be used as the CTE ## ..auto.. adjust the maxDate if (maxDate == "..auto..") { max_from_table <- qMaxDate(tbl=tbl_in, schema=schema_in) wdays <- getWdays() ## If we have less than half 3/7ths of the week, filter the maxDate, otherwise, no filter if (wdays[wday(max_from_table)] %ni% c("Fri", "Sat", "Sun")) maxDate <- friday_weekof(max_from_table) %>% unname else maxDate <- NULL } qry.counts <- makeQry(tbl=tbl_in, schema=schema_in, colsToPull=c(colsToPull,colsWithaggFunc), colsToAgg=colsToAgg, aggFunc="count", dateCol=dateCol_in, maxDate=maxDate, limit=NULL) ## The WITH ... CTE clause ## This will actually be applied in the makeQry() call for the final, ie the cumsum, query ## However, it's name will be referenced in the third, ie the self_join, query with.cumsum <- c(CTE_Counts=qry.counts) ## ------ SELF_JOIN QUERY --------------------------------- ## Will agg the original columns, plus aggregate new_users ## Preface colsToAgg with T2, and colsToPull with T1 ## The columns that will be cumsumd are "new_users", and the colsToAgg, assuming the aggFunc is itself sum ## (in the default, the colsToAgg will simply be "streams" and the aggFunc will be "sum") colsThatWillBeCumSumd <- colsToAgg %>% colNamesFromVector %>% {setNames(obj=paste0("T2.", .), nm=paste0("T2_", .))} ## The colsToPull aka, the grouping columns in the final query, will be the colsToPull from previous, ## plus any cols we previously aggregated (generally thats "new_users" and "streams") colsToPull.self_join <- c(colsToPull, colsWithaggFunc, colsToAgg) %>% colNamesFromVector # %>% paste0("T1.", .) ## The main part of the "ON" clause is the T1.date >= T2.date, which appears below ## However, in addtion to this, we must also join on each group ## Since NULL values are possible in any group, we need to account for them with equal_null(X, Y) ## ----- ## HOWEVER, due to the snowflake bug in equal_null(), we have to get creative and use NVL() ## which requires us to know the data type of each column ## --------- ## The columns to join on are the group columns (from counts query), less the date column colsToJoinOn <- colsToPull %>% colNamesFromVector %>% setdiff(dateCol.nm) ## Create the additional ON_AND clause; If there are no such columns, it should be blank (hence ifelse()) ON_AND.self_join <- ifelse (!length(colsToJoinOn), "", { # colsToPull.self_join %>% setdiff("date") %>% sprintf("equal_null(T1.%s, T2.%1$s)", .) %>% paste(" AND ", .) %>% pasteC ## DOESN'T WORK. BUGGY ## ------------------------------------------------- ## frmt.null_joiner <- "equal_null(T1.%s, T2.%1$s)" ## equal_null() is buggy, thus using NVL() colsToJoinOn %>% sprintf(frmt.null_joiner, .) %>% paste("\n AND ", .) %>% pasteC ## ------------------------------------------------- ## ## TEMP FIX ## ------------------------------------------------- ## frmt.null_joiner.numeric <- "NVL(T1.%s, -123.1230321) = NVL(T2.%1$s, -123.1230321)" frmt.null_joiner.string <- "NVL(T1.%s, 'dummy_dummy_value___xyz123') = NVL(T2.%1$s, 'dummy_dummy_value___xyz123')" ## Which columns to use the numeric version. Technically UPC is numeric and album_code is string; but should only use album_code anyway tmp.known_numeric_cols <- c("playlist_is_unknown", "user_birthyear", "length_seconds", "row_id") colsToJoinOn %>% {sprintf(ifelse(. %in% tmp.known_numeric_cols, frmt.null_joiner.numeric, frmt.null_joiner.string), .)} %>% paste("\n AND ", .) %>% pasteC() ## ------------------------------------------------- ## }) ## Cobine the three pieces of the join statement ## The acutal join clause; the date filter; the ON_AND for the rest of the group columns join_statement.self_join <- sprintf("\n INNER JOIN %s T2\n ON T1.%s >= T2.%s %s", names(with.cumsum), dateCol.nm, dateCol.nm, ON_AND.self_join) ## The table is the name of the CTE from the WITH clause tbl.self_join <- c(T1=names(with.cumsum)) qry.self_join <- makeQry(with=NULL, tbl=tbl.self_join, colsToPull=colsToPull.self_join %>% paste0("T1.", .) %>% c(colsThatWillBeCumSumd), colsToAgg=NULL, join=join_statement.self_join, aggFunc=NULL, prependCols.with.tbl=FALSE, nogroupby=TRUE, limit=NULL) ## ------ CUMULATIVE SUM QUERY --------------------------------- ## The cumsum query is pulling the same columns as the self_join query, less the T2 columns, which will instead be aggregated colsToPull.cumsum <- colsToPull.self_join ## Grouping by the same cols. We want to be explicit here. We might be adding in the addl_cols, which makeQry() will otherwise add these to the Group By clause and there might be a loss of efficiency groupby.cumsum <- colsToPull.self_join ## Agg the colsThatWillBeCumSumd colsToAgg.cumsum <- colsThatWillBeCumSumd %>% colNamesFromVector() %>% {setNames(., nm=gsub("T2", "cumulative", .))} ## CREATE AN ACCELERATION SCORE ## - For now, Only do so if the only colsToAgg is "streams" ## This could be expanded to other colsToAgg if needed, so long as the aggFunc is "SUM" if (length(colsToAgg)) { cols <- colsToAgg %>% colNamesFromVector() addl_cols <- lapply(unname(cols), make_accel_cols, orderby=dateCol.nm, partitionby=colsToPull.cumsum) %>% unlist(recursive=FALSE, use.names=TRUE) } else { addl_cols <- NULL } qry.cumsum <- makeQry(with=with.cumsum, tbl=qry.self_join, colsToPull=c(colsToPull.cumsum, addl_cols), colsToAgg=colsToAgg.cumsum, aggFunc="sum", prependCols.with.tbl=FALSE, groupby=groupby.cumsum, limit=NULL) # if (verbose.qry) # catn(verboseQry(qry.cumsum, max.width=150, max.lines=55, add.dots=FALSE)) browser(expr=inDebugMode(c("stream_counts", "create_spotify_stream_counts_table")), text="in create_spotify_stream_counts_table() right before calling sfPopulateTable()") ## catn(verboseQry(qry.cumsum)) sfPopulateTable(tbl=tbl_out, schema=schema, qry=qry.cumsum, over=TRUE, verbose.rows=TRUE, verbose=verbose.qry) if (confirm_streams) { tryCatch({ streams_users <- makeQry(tbl=tbl_out, schema=schema, colsToAgg=c("streams"), aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist() streams <- streams_users[["streams"]] msg <- sprintf("Total number of streams %s (%s).", ifelse(streams == streams_expected, "confirmed", "is *WRONG*"), formnumb(streams, round=FALSE)) verboseMsg(TRUE, msg, func="message", minw=100) }, error=function(e) { warning("Was unable to confirm streams due to an error. The error was caught and execution unaffected. Caught error message: \n\t", e$message, "\n", call.=FALSE); }) } return(tbl_out) }