Revise export to batch based on row count

R/Database.R

@@ -318,43 +318,96 @@ exportDatabaseToCsv <- function(
 
   # extract result per table
   for(table in tables){
-    message(paste0("Exporting ", table))
-    sql <- "select * from @resultSchema.@appendtotable@tablename;"
+    ParallelLogger::logInfo(paste0('Exporting rows from ', table, ' to csv file'))
+    # get row count and figure out number of loops
+    sql <- "select count(*) as N from @resultSchema.@appendtotable@tablename;"
     sql <- SqlRender::render(
       sql = sql,
       resultSchema = resultSchema,
       appendtotable = tablePrefix,
       tablename = table
     )
-    resultSet <- DatabaseConnector::dbSendQuery(connection, sql)
-    tryCatch({
-      i <- 1
-      while (i == 1 || !DatabaseConnector::dbHasCompleted(resultSet)) {
-        start <- format(x = (i-1)*maxRowCount+1, scientific = F, big.mark = ",")
-        end <- format(x = maxRowCount*i, scientific = F, big.mark = ",")
-        message(paste0("  -- Rows ", start, " to ", end))
-        result <- DatabaseConnector::dbFetch(resultSet, n = maxRowCount)
-        if (table == "covariates" && minMeanCovariateValue > 0) {
-          result <- result %>%
-            dplyr::filter(.data$average_value >= minMeanCovariateValue)
-        }
-        result <- formatDouble(result)
-        # save the results as a csv
-        readr::write_csv(
-          x = result,
-          file = file.path(saveDirectory, paste0(tolower(filePrefix), table,'.csv')),
-          append = (i > 1)
-        )
-        i <- i + 1
-      }
-    },
-    error = function(e) {
-      message(paste0("ERROR in export to csv: ", e$message));
-    },
-    finally = {
-      DatabaseConnector::dbClearResult(resultSet)
-    })
+    sql <- SqlRender::translate(
+      sql = sql,
+      targetDialect = targetDialect,
+      tempEmulationSchema = tempEmulationSchema
+    )
+    countN <- DatabaseConnector::querySql(
+      connection = connection,
+      sql = sql,
+      snakeCaseToCamelCase = F
+    )$N
+
+    # get column names
+    sql <- "select * from @resultSchema.@appendtotable@tablename where 1=0;"
+    sql <- SqlRender::render(
+      sql = sql,
+      resultSchema = resultSchema,
+      appendtotable = tablePrefix,
+      tablename = table
+    )
+    sql <- SqlRender::translate(
+      sql = sql,
+      targetDialect = targetDialect,
+      tempEmulationSchema = tempEmulationSchema
+    )
+    cnames <- colnames(DatabaseConnector::querySql(
+      connection = connection,
+      sql = sql,
+      snakeCaseToCamelCase = F
+    ))
+
+    inds <- floor(countN/maxRowCount)
+    tableAppend = F
+    pb = utils::txtProgressBar(min = 0, max = countN, initial = 0)
+
+    for(i in 1:length(inds)){
+
+      startRow <- (i-1)*maxRowCount + 1
+      endRow <- min(i*maxRowCount, countN)
+
+      sql <- "select @cnames from
+    (select *,
+    ROW_NUMBER() OVER(ORDER BY @cnames) AS row
+    from @resultSchema.@appendtotable@tablename
+    ) temp
+    where
+    temp.row >= @start_row and
+    temp.row <= @end_row;"
+      sql <- SqlRender::render(
+        sql = sql,
+        resultSchema = resultSchema,
+        appendtotable = tablePrefix,
+        tablename = table,
+        cnames = paste(cnames, collapse = ','),
+        start_row = startRow,
+        end_row = endRow
+      )
+      sql <- SqlRender::translate(
+        sql = sql,
+        targetDialect = targetDialect,
+        tempEmulationSchema = tempEmulationSchema
+      )
+      result <- DatabaseConnector::querySql(
+        connection = connection,
+        sql = sql,
+        snakeCaseToCamelCase = F
+      )
+      result <- formatDouble(result)
+
+      # save the results as a csv
+      readr::write_csv(
+        x = result,
+        file = file.path(saveDirectory, paste0(tolower(filePrefix), table,'.csv')),
+        append = tableAppend
+      )
+      tableAppend = T
+      utils::setTxtProgressBar(pb,endRow)
+
+    }
+    close(pb)
   }
+
   invisible(saveDirectory)
 }