Bring data into R before doing a big left join -- Acero has a bug res…

…ulting in a limit of 4GB on the right table (apache/arrow#37655)

NAMESPACE

@@ -17,6 +17,7 @@ export(setnafill)
 export(setunite)
 export(stream_debounce)
 export(tessi_changed_emails)
+importFrom(arrow,as_arrow_table)
 importFrom(arrow,concat_tables)
 importFrom(censusxy,cxy_geography)
 importFrom(checkmate,assert)

R/email_stream.R

@@ -183,10 +183,10 @@ email_fix_eaddress <- function(email_stream) {
   setleftjoin(email_matches, emails_cleaned, by = "eaddress")
 
   # provide event_type and event_subtype and remove columns
-  email_stream <- email_stream %>% select(-eaddress) %>% compute %>%
+  email_stream <- email_stream %>% select(-eaddress) %>% collect %>%
     left_join(email_matches[,.(customer_no,timestamp_id,eaddress,domain)],
               by = c("customer_no","timestamp_id")) %>%
-    compute
+    as_arrow_table
 
 }
 
@@ -289,8 +289,8 @@ email_stream_base <- function(from_date = as.POSIXct("1900-01-01"), to_date = no
     return(arrow::arrow_table(group_customer_no=integer(0)))
   }
 
-  email_stream %>% email_data_append(...) %>% email_fix_timestamp %>% email_fix_eaddress %>%
-  filter(timestamp >= from_date & timestamp < to_date) %>%
+  email_stream %>% email_data_append(...) %>% email_fix_timestamp %>%
+  filter(timestamp >= from_date & timestamp < to_date) %>% email_fix_eaddress %>%
   transmute(group_customer_no = as.integer(group_customer_no), customer_no,
             timestamp, timestamp_id, event_type = "Email", event_subtype,
             source_no, appeal_no, campaign_no, source_desc, extraction_desc,

tests/testthat/test-email_stream.R

@@ -74,7 +74,7 @@ test_that("email_fix_timestamp recalculates send timestamps based on earliest pr
 test_that("email_fix_eaddress fills in customer emails based on the send date", {
   email_fix_timestamp <- email_data_stubbed() %>% email_data_append_stubbed() %>%
     email_fix_timestamp %>% collect %>% setDT
-  email_fix_eaddress <- email_fix_eaddress_stubbed(email_fix_timestamp)
+  email_fix_eaddress <- email_fix_eaddress_stubbed(email_fix_timestamp) %>% collect %>% setDT
   emails <- readRDS(rprojroot::find_testthat_root_file("email_stream-emails.Rds")) %>% setDT %>%
     setkey(customer_no, timestamp)