Basic functionality and tests for python csv uploads

.github/workflows/R_CMD_check_Hades.yml

@@ -52,6 +52,7 @@ jobs:
       CDM5_SPARK_PASSWORD: ${{ secrets.CDM5_SPARK_PASSWORD }}
       CDM5_SPARK_CONNECTION_STRING: ${{ secrets.CDM5_SPARK_CONNECTION_STRING }}
 
+
     steps:
       - uses: actions/checkout@v2
 
@@ -75,9 +76,29 @@ jobs:
 
       - uses: r-lib/actions/setup-r-dependencies@v2
         with:
-          extra-packages: any::rcmdcheck
+          extra-packages: any::rcmdcheck reticulate
           needs: check
 
+
+      - uses: actions/setup-python@v4
+        with:
+          python-version: "3.x"
+
+      - name: setup r-reticulate venv
+        shell: Rscript {0}
+        run: |
+
+          path_to_python <- reticulate::virtualenv_create(
+            envname = "r-reticulate",
+            python = Sys.which("python"), # placed on PATH by the setup-python action
+            packages = c(
+              "psycopg2-binary"
+            )
+          )
+
+          writeLines(sprintf("RETICULATE_PYTHON=%s", path_to_python),
+                     Sys.getenv("GITHUB_ENV"))
+
       - uses: r-lib/actions/check-r-package@v2
         with:
           args: 'c("--no-manual", "--as-cran")'

.github/workflows/R_CMD_check_main_weekly.yaml

@@ -59,6 +59,20 @@ jobs:
 
       - uses: r-lib/actions/setup-pandoc@v2
 
+      - name: setup r-reticulate venv
+        shell: Rscript {0}
+        run: |
+
+          path_to_python <- reticulate::virtualenv_create(
+            envname = "r-reticulate",
+            python = Sys.which("python"), # placed on PATH by the setup-python action
+            packages = c(
+              "psycopg2-binary"
+            )
+          )
+
+          writeLines(sprintf("RETICULATE_PYTHON=%s", path_to_python),
+                     Sys.getenv("GITHUB_ENV"))
       - uses: r-lib/actions/setup-r-dependencies@v2
         with:
           extra-packages: any::rcmdcheck

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: ResultModelManager
 Title: Result Model Manager
-Version: 0.5.11
+Version: 0.6.0
 Authors@R:
     person("Jamie", "Gilbert", , "gilbert@ohdsi.org", role = c("aut", "cre"))
 Description: Database data model management utilities for R packages in the Observational Health Data Sciences and
@@ -42,5 +42,7 @@ Suggests:
   pkgdown,
   remotes,
   styler,
-  Andromeda
+  Andromeda,
+  rJava,
+  reticulate
 Config/testthat/edition: 3

NAMESPACE

@@ -9,9 +9,15 @@ export(createQueryNamespace)
 export(createResultExportManager)
 export(deleteAllRowsForDatabaseId)
 export(deleteAllRowsForPrimaryKey)
+export(disablePythonUploads)
+export(enablePythonUploads)
 export(generateSqlSchema)
 export(grantTablePermissions)
+export(install_psycopg2)
 export(loadResultsDataModelSpecifications)
+export(pyPgUploadEnabled)
+export(pyUploadCsv)
+export(pyUploadDataFrame)
 export(unzipResults)
 export(uploadResults)
 import(DatabaseConnector)

NEWS.md

@@ -1,3 +1,9 @@
+# ResultModelManager 0.6.0
+
+Changes:
+1. Added optional wrapper functions for python based 
+
+
 # ResultModelManager 0.5.11
 
 Changes:

R/DataModel.R

@@ -48,7 +48,8 @@ checkAndFixColumnNames <-
 
     expectedNames <- tableSpecs %>%
       dplyr::select("columnName") %>%
-      dplyr::anti_join(dplyr::filter(optionalNames, !.data$columnName %in% observeredNames),
+      dplyr::anti_join(
+        dplyr::filter(optionalNames, !.data$columnName %in% observeredNames),
         by = "columnName"
       ) %>%
       dplyr::arrange("columnName") %>%
@@ -181,7 +182,7 @@ checkAndFixDuplicateRows <-
            specifications) {
     primaryKeys <- specifications %>%
       dplyr::filter(.data$tableName == !!tableName &
-        tolower(.data$primaryKey) == "yes") %>%
+                      tolower(.data$primaryKey) == "yes") %>%
       dplyr::select("columnName") %>%
       dplyr::pull()
     duplicatedRows <- duplicated(table[, primaryKeys])
@@ -194,7 +195,7 @@ checkAndFixDuplicateRows <-
           sum(duplicatedRows)
         )
       )
-      return(table[!duplicatedRows, ])
+      return(table[!duplicatedRows,])
     } else {
       return(table)
     }
@@ -220,7 +221,7 @@ appendNewRows <-
     if (nrow(data) > 0) {
       primaryKeys <- specifications %>%
         dplyr::filter(.data$tableName == !!tableName &
-          tolower(.data$primaryKey) == "yes") %>%
+                        tolower(.data$primaryKey) == "yes") %>%
         dplyr::select("columnName") %>%
         dplyr::pull()
       newData <- newData %>%
@@ -250,10 +251,10 @@ formatDouble <- function(x) {
 
 .truncateTable <- function(tableName, connection, schema, tablePrefix) {
   DatabaseConnector::renderTranslateExecuteSql(connection,
-    "TRUNCATE TABLE @schema.@table_prefix@table;",
-    table_prefix = tablePrefix,
-    schema = schema,
-    table = tableName
+                                               "TRUNCATE TABLE @schema.@table_prefix@table;",
+                                               table_prefix = tablePrefix,
+                                               schema = schema,
+                                               table = tableName
   )
   invisible(NULL)
 }
@@ -273,7 +274,7 @@ formatDouble <- function(x) {
 }
 
 
-uploadChunk <- function(chunk, pos, env, specifications, resultsFolder, connection, runCheckAndFixCommands, forceOverWriteOfSpecifications) {
+uploadChunk <- function(chunk, pos, env, specifications, resultsFolder, connection, pythonConnection, runCheckAndFixCommands, forceOverWriteOfSpecifications) {
   ParallelLogger::logInfo(
     "- Preparing to upload rows ",
     pos,
@@ -354,8 +355,8 @@ uploadChunk <- function(chunk, pos, env, specifications, resultsFolder, connecti
     primaryKeyValuesInChunk <- unique(chunk[env$primaryKey])
     duplicates <-
       dplyr::inner_join(env$primaryKeyValuesInDb,
-        primaryKeyValuesInChunk,
-        by = env$primaryKey
+                        primaryKeyValuesInChunk,
+                        by = env$primaryKey
       )
 
     if (nrow(duplicates) != 0) {
@@ -386,23 +387,33 @@ uploadChunk <- function(chunk, pos, env, specifications, resultsFolder, connecti
       # Remove duplicates we already dealt with:
       env$primaryKeyValuesInDb <-
         env$primaryKeyValuesInDb %>%
-        dplyr::anti_join(duplicates, by = env$primaryKey)
+          dplyr::anti_join(duplicates, by = env$primaryKey)
     }
   }
   if (nrow(chunk) == 0) {
     ParallelLogger::logInfo("- No data left to insert")
   } else {
     insertTableStatus <- tryCatch(expr = {
-      DatabaseConnector::insertTable(
-        connection = connection,
-        tableName = env$tableName,
-        databaseSchema = env$schema,
-        data = chunk,
-        dropTableIfExists = FALSE,
-        createTable = FALSE,
-        tempTable = FALSE,
-        progressBar = TRUE
-      )
+      if (!is.null(pythonConnection)) {
+        tryCatch({
+          .pgWriteDataFrame(chunk, pyConnection = pythonConnection, table = env$tableName, schema = env$schema)
+        }, error = function(error) {
+          # rollback write of data
+          pythonConnection$rollback()
+          stop(error)
+        })
+      } else {
+        DatabaseConnector::insertTable(
+          connection = connection,
+          tableName = env$tableName,
+          databaseSchema = env$schema,
+          data = chunk,
+          dropTableIfExists = FALSE,
+          createTable = FALSE,
+          tempTable = FALSE,
+          progressBar = TRUE
+        )
+      }
     }, error = function(e) e)
     if (inherits(insertTableStatus, "error")) {
       stop(insertTableStatus$message)
@@ -421,7 +432,8 @@ uploadTable <- function(tableName,
                         runCheckAndFixCommands,
                         forceOverWriteOfSpecifications,
                         purgeSiteDataBeforeUploading,
-                        warnOnMissingTable) {
+                        warnOnMissingTable,
+                        pythonConnection) {
   csvFileName <- paste0(tableName, ".csv")
   specifications <- specifications %>%
     dplyr::filter(.data$tableName == !!tableName)
@@ -485,11 +497,11 @@ uploadTable <- function(tableName,
     convertType <- Vectorize(
       function(type) {
         switch(type,
-          varchar = "c",
-          bigint = "n",
-          int = "n",
-          date = "D",
-          "?"
+               varchar = "c",
+               bigint = "n",
+               int = "n",
+               date = "D",
+               "?"
         ) # default to guess if type not matched
       }
     )
@@ -501,7 +513,7 @@ uploadTable <- function(tableName,
 
     readr::read_csv_chunked(
       file = file.path(resultsFolder, csvFileName),
-      callback = function(chunk, pos) uploadChunk(chunk, pos, env, specifications, resultsFolder, connection, runCheckAndFixCommands, forceOverWriteOfSpecifications),
+      callback = function(chunk, pos) uploadChunk(chunk, pos, env, specifications, resultsFolder, connection, pythonConnection, runCheckAndFixCommands, forceOverWriteOfSpecifications),
       chunk_size = 1e7,
       col_types = colTypes,
       guess_max = 1e6,
@@ -593,10 +605,10 @@ uploadResults <- function(connection = NULL,
     ParallelLogger::logInfo("Removing all records for tables within specification")
 
     invisible(lapply(unique(specifications$tableName),
-      .truncateTable,
-      connection = connection,
-      schema = schema,
-      tablePrefix = tablePrefix
+                     .truncateTable,
+                     connection = connection,
+                     schema = schema,
+                     tablePrefix = tablePrefix
     ))
   }
 
@@ -625,7 +637,11 @@ uploadResults <- function(connection = NULL,
       )
     }
   }
-
+  pythonConnection <- NULL
+  if (DatabaseConnector::dbms(connection) == "postgresql" && pyPgUploadEnabled()) {
+    pythonConnection <- .createPyConnection(connection)
+    on.exit(pythonConnection$close(), add = TRUE)
+  }
   for (tableName in unique(specifications$tableName)) {
     uploadTable(
       tableName,
@@ -638,8 +654,13 @@ uploadResults <- function(connection = NULL,
       runCheckAndFixCommands,
       forceOverWriteOfSpecifications,
       purgeSiteDataBeforeUploading,
-      warnOnMissingTable
+      warnOnMissingTable,
+      pythonConnection = pythonConnection
     )
+
+    if (!is.null(pythonConnection)) {
+      pythonConnection$commit()
+    }
   }
 
   delta <- Sys.time() - start
@@ -660,6 +681,7 @@ uploadResults <- function(connection = NULL,
 #' @export
 deleteAllRowsForPrimaryKey <-
   function(connection, schema, tableName, keyValues) {
+
     createSqlStatement <- function(i) {
       sql <- paste0(
         "DELETE FROM ",
@@ -668,7 +690,7 @@ deleteAllRowsForPrimaryKey <-
         tableName,
         "\nWHERE ",
         paste(paste0(
-          colnames(keyValues), " = '", keyValues[i, ], "'"
+          colnames(keyValues), " = '", keyValues[i,], "'"
         ), collapse = " AND "),
         ";"
       )
@@ -743,9 +765,9 @@ deleteAllRowsForDatabaseId <-
         database_id = databaseId
       )
       DatabaseConnector::executeSql(connection,
-        sql,
-        progressBar = FALSE,
-        reportOverallTime = FALSE
+                                    sql,
+                                    progressBar = FALSE,
+                                    reportOverallTime = FALSE
       )
     }
   }