fix: 🐛 use duckplyr to read Parquet as DuckDB

DESCRIPTION

@@ -20,6 +20,7 @@ BugReports: https://github.com/dp-next/fastreg/issues
 Depends:
     R (>= 4.1.0)
 Imports:
+    duckplyr (>= 1.2.1),
     arrow,
     checkmate,
     cli,
@@ -46,7 +47,8 @@ Suggests:
     testthat (>= 3.0.0),
     tidyselect,
     withr,
-    tarchetypes
+    tarchetypes,
+    duckdb (>= 1.5.4)
 VignetteBuilder:
     quarto
 Encoding: UTF-8

R/read.R

@@ -57,13 +57,13 @@ read_parquet_dataset <- function(path) {
   checkmate::assert_directory_exists(path)
   assert_directory_not_empty(path)
   path |>
-    arrow::open_dataset(
-      unify_schemas = TRUE,
-      # Explicitly set type of partition to int32 to handle when year is
-      # missing.
-      partitioning = arrow::hive_partition(year = arrow::int32())
-    ) |>
-    arrow::to_duckdb()
+    duckplyr::read_parquet_duckdb(
+      prudence = "thrifty",
+      options = list(
+        union_by_name = TRUE,
+        hive_partitioning = TRUE
+      )
+    )
 }
 
 #' @describeIn read_parquet Reads a single Parquet file.
@@ -73,8 +73,9 @@ read_parquet_file <- function(path) {
   checkmate::assert_file_exists(path)
   assert_is_parquet(path)
   path |>
-    arrow::read_parquet() |>
-    arrow::to_duckdb()
+    duckplyr::read_parquet_duckdb(
+      prudence = "thrifty"
+    )
 }
 
 assert_is_parquet <- function(path) {

README.qmd

@@ -10,9 +10,11 @@ gh:
 <!-- badges: start -->
 
 [![CRAN status](https://www.r-pkg.org/badges/version/fastreg)](https://CRAN.R-project.org/package=fastreg)
-[![GitHub Release](https://img.shields.io/github/v/release/{{< meta gh.org >}}/{{< meta gh.repo >}}.svg)](https://github.com/{{< meta gh.org >}}/{{< meta gh.repo >}}/releases/latest)
+[![GitHub
+Release](https://img.shields.io/github/v/release/{{< meta gh.org >}}/{{< meta gh.repo >}}.svg)](https://github.com/{{< meta gh.org >}}/{{< meta gh.repo >}}/releases/latest)
 [![Build](https://github.com/{{< meta gh.org >}}/{{< meta gh.repo >}}/actions/workflows/build.yml/badge.svg)](https://github.com/{{< meta gh.org >}}/{{< meta gh.repo >}}/actions/workflows/build.yml)
-[![pre-commit.ci status](https://results.pre-commit.ci/badge/github/{{< meta gh.org >}}/{{< meta gh.repo >}}/main.svg)](https://results.pre-commit.ci/latest/github/{{< meta gh.org >}}/{{< meta gh.repo >}}/main)
+[![pre-commit.ci
+status](https://results.pre-commit.ci/badge/github/{{< meta gh.org >}}/{{< meta gh.repo >}}/main.svg)](https://results.pre-commit.ci/latest/github/{{< meta gh.org >}}/{{< meta gh.repo >}}/main)
 [![lifecycle](https://lifecycle.r-lib.org/articles/figures/lifecycle-experimental.svg)](https://lifecycle.r-lib.org/articles/stages.html#experimental)
 [![Project Status: Active – The project has reached a stable, usable state and is being actively developed.](https://www.repostatus.org/badges/latest/active.svg)](https://www.repostatus.org/#active)
 

tests/testthat/test-read.R

@@ -12,6 +12,8 @@ purrr::walk(sas_paths$output_path, \(path) {
 # Test read_parquet_*() ---------------------------------------------------------
 
 test_that("read_parquet_file() reads a single Parquet file", {
+  gc() # Force any previous connection to be dropped to avoid "View with name X already exists"-errors during tests
+
   # Read single Parquet file (2020 file).
   # Because UUID is used in the convert function, we can't know the name of the
   # file.
@@ -27,8 +29,8 @@ test_that("read_parquet_file() reads a single Parquet file", {
   expected_data <- haven::read_sas(expected_source_file)
 
   expect_equal(
-    # year col doesn't exist when only one file is read.
-    actual_data |> dplyr::select(-"source_file"),
+    # source file and year columns don't exist in the basic haven conversion
+    actual_data |> dplyr::select(-c("source_file", "year")),
     expected_data,
     ignore_attr = TRUE
   )
@@ -39,6 +41,7 @@ test_that("read_parquet_file() reads a single Parquet file", {
 })
 
 test_that("read_parquet_dataset() reads a partitioned Parquet register", {
+  gc() # Force any previous connection to be dropped to avoid "View with name X already exists"-errors during tests
   actual <- read_parquet_dataset(output_dir) |> dplyr::collect()
 
   expected <- purrr::map(sas_paths$output_path, \(path) {
@@ -106,6 +109,7 @@ test_that("read_parquet_file() errors when file is not Parquet", {
 })
 
 test_that("files with extension .parq can also be read", {
+  gc() # Force any previous connection to be dropped to avoid "View with name X already exists"-errors during tests
   path <- fs::path_temp("file.parq")
   arrow::write_parquet(
     simulate_registers_with_paths("bef")$data[[1]],
@@ -114,8 +118,9 @@ test_that("files with extension .parq can also be read", {
   expect_no_error(read_parquet_file(path))
 })
 
-
 test_that("read_parquet_dataset() reads files with different columns", {
+  gc() # Force any previous connection to be dropped to avoid "View with name X already exists"-errors during tests
+
   # Faux bef with lmdb structure, saved separately and combined with sas paths
   sas_dir <- fs::path_temp("different_columns/sas")
   parquet_dir <- fs::path_temp("different_columns/parquet")
@@ -147,33 +152,11 @@ test_that("read_parquet_dataset() reads files with different columns", {
   )
 })
 
-test_that("read_parquet_dataset() errors with incompatible schemas", {
-  # Create a bef file where numeric columns are changed to character, so
-  # the schema is incompatible with the other bef files.
-  incompatible_data <- bef_list$data[[1]] |>
-    dplyr::mutate(dplyr::across(where(is.numeric), as.character))
-
-  incompatible_sas_path <- fs::path_temp(
-    "sas_schema_incompatible/bef2099.sas7bdat"
-  )
-  write_to_sas(incompatible_data, incompatible_sas_path)
-  sas_incompatible <- c(
-    sas_paths$output_path,
-    incompatible_sas_path
-  )
-
-  incompatible_output <- fs::path_temp("incompatible")
-  # Convert files.
-  purrr::walk(sas_incompatible, \(path) {
-    convert(path, incompatible_output)
-  })
-
-  expect_error(read_parquet_dataset(incompatible_output), "incompatible")
-})
-
 # Test read_register() ---------------------------------------------------------
 
 test_that("read in a register", {
+  gc() # Force any previous connection to be dropped to avoid "View with name X already exists"-errors during tests
+
   withr::with_options(
     list(
       fastreg.project_rawdata_dir = fs::path_temp("E/rawdata/202020/"),
@@ -194,7 +177,7 @@ test_that("read in a register", {
         get_project_workdata_dir()
       )
 
-      expect_match(class(read_register("bef")), "duckdb", all = FALSE)
+      expect_match(class(read_register("bef")), "duckplyr_df", all = FALSE)
       expect_shape(dplyr::collect(read_register("bef")), nrow = 10)
       expect_error(
         # Warning about project id from internal `get_project_rawdata_dir()`.

tests/testthat/test-use.R

@@ -91,17 +91,35 @@ test_that("targets pipeline template converts SAS files to Parquet", {
     purrr::list_c() |>
     sum()
 
-  n_actual_bef <- arrow::open_dataset(fs::path(
+  n_actual_bef <- fs::path(
     test_output_dir,
     "bef"
-  )) |>
+  ) |>
+    duckplyr::read_parquet_duckdb(
+      prudence = "thrifty",
+      options = list(
+        union_by_name = TRUE,
+        hive_partitioning = TRUE
+      )
+    ) |>
+    duckplyr::as_duckdb_tibble() |>
+    duckplyr::as_tbl() |>
     dplyr::collect() |>
     nrow()
 
-  n_actual_lmdb <- arrow::open_dataset(fs::path(
+  n_actual_lmdb <- fs::path(
     test_output_dir,
     "lmdb"
-  )) |>
+  ) |>
+    duckplyr::read_parquet_duckdb(
+      prudence = "thrifty",
+      options = list(
+        union_by_name = TRUE,
+        hive_partitioning = TRUE
+      )
+    ) |>
+    duckplyr::as_duckdb_tibble() |>
+    duckplyr::as_tbl() |>
     dplyr::collect() |>
     nrow()