R/update_data.r

In giupo/GrafoDB: simple ETL for timeseries in R

#' @include conflicts.r

update_data <- function(x, con, tag = x@tag, notes = "") {
  ln <- "GrafoDB::update_data"

  if (interactive()) rutils::.info("Update Data ...", name = ln)

  data <- x@data
  autore <- rutils::whoami()

  df <- if (length(hash::keys(data))) {
    DBI::dbGetQuery(con, sql_by_key(
      "GET_CHANGED_DATA",
      tag = tag,
      last_updated = as.numeric(x@timestamp),
      .con = con))
  } else {
    data.frame()
  }

  names_with_conflicts <- intersect(x@touched, as.character(df$name))
  names_updated <- setdiff(hash::keys(data), names_with_conflicts)

  if (length(names_updated)) {
    stage_name <- "stage"
    # create temporary data with names_updated

    # creo il data.frame dei dati da usare nel DB
    tryCatch({
      last_updated <- time_in_nano()
      name <- NULL
      dati <- foreach::`%dopar%`(foreach::foreach(
        name = iterators::iter(names_updated), .combine = rbind), {
        df <-  to_data_frame(data[[name]])
        cbind(df, name, tag, autore, notes, last_updated)
      }) # this is quite fast, let's ignore the Progressbar here...

      DBI::dbExecute(con, sql_by_key(
        "CREATE_STAGE", stage_name = stage_name,
        .con = con))

      on.exit({
        tryCatch({
          DBI::dbExecute(con,
            sql_by_key("DROP_STAGE", stage_name = stage_name, .con = con))
        }, error = function(cond) {
          rutils::.warn(cond, name = ln)
        })
      })

      DBI::dbWriteTable(con, stage_name, dati,
        row.names = FALSE, overwrite = TRUE)

      # aggiorna i record esistenti...
      DBI::dbExecute(con, sql_by_key(
        "UPDATE_WITH_STAGE",
        tag = tag,
        stage_name = stage_name,
        .con = con))

      DBI::dbExecute(con, sql_by_key(
        "INSERT_WITH_STAGE",
        tag = tag,
        stage_name = stage_name,
        .con = con))
    }, error = function(cond) {
      stop(cond)
    })
  }
  if (interactive()) rutils::.info("Update Data done.", name = ln)
}