prep_open_dataset <- function(path_prep, partitioning = "sboe_id", ...) {
  if (!fs::file_exists(path_prep)) {
    path_here <- here::here("data-prep/", path_prep)
    path_up <- fs::path("..", "data-prep", path_prep)
    if (fs::file_exists(path_here)) {
      path_prep <- path_here
    } else if (fs::file_exists(path_up)) {
      path_prep <- path_up
    } else {
      stop("File not found: ", 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)
}