prep_open_dataset <- function(path_prep, partitioning = "sboe_id", ...) {
  path_prep <- resolve_path_data_prep(path_prep)

  if (length(fs::dir_ls(path_prep, type = "dir")) == 0) {
    partitioning <- NULL
  }

  arrow::open_dataset(path_prep, partitioning = partitioning, ...)
}

prep_open_dataset_db <- function(table, ..., path_prep = table) {
  pq <- prep_open_dataset(path_prep, ...)

  con <- duckdb_global_con()
  duckdb::duckdb_register_arrow(con, table, pq)
  dplyr::tbl(con, table)
}

prep_open_address_db <- function(
  path_db = "address_lookup.sqlite"
) {
  path_db <- resolve_path_data_prep(path_db)

  con <- if (!is.null(.globals$con_address)) {
    .globals$con_address
  } else {
    .globals$con_address <- DBI::dbConnect(RSQLite::SQLite(), path_db)
  }

  tbl(con, "resolved")
}

# Utils ----
resolve_path_data_prep <- function(path_prep) {
  if (fs::file_exists(path_prep)) {
    return(path_prep)
  }

  path_here <- here::here("data-prep/", path_prep)
  path_up <- fs::path("..", "data-prep", path_prep)
  path_up2 <- fs::path("..", "..", "data-prep", path_prep)

  if (fs::file_exists(path_here)) {
    return(path_here)
  }

  if (fs::file_exists(path_up)) {
    return(path_up)
  }

  if (fs::file_exists(path_up2)) {
    return(path_up2)
  }

   stop("File not found: ", path_prep)
}