Databases and dbplyr

dbplyr is the reason R belongs in a data pipeline. The same dplyr code that ran on a tibble in lesson 2 runs against a warehouse table, translated to SQL, with the computation happening there and only the answer coming back.

Connecting

suppressPackageStartupMessages({library(tidyverse); library(DBI); library(duckdb); library(dbplyr)})

con <- dbConnect(duckdb(), dbdir = "bookshop.duckdb")

set.seed(42)
n <- 2e6
dbWriteTable(con, "orders", tibble(
  order_id    = 1:n,
  customer_id = sample(1:50000, n, TRUE),
  ordered_at  = as.Date("2026-01-01") + sample(0:89, n, TRUE),
  channel     = sample(c("web","app","phone"), n, TRUE),
  status      = sample(c("completed","returned","refunded","pending"), n, TRUE),
  amount      = round(rlnorm(n, 3.0, 0.65), 2)
), overwrite = TRUE)

dbWriteTable(con, "customers", tibble(
  customer_id = 1:50000,
  country     = sample(c("GB","US","NL","DE"), 50000, TRUE),
  signed_up   = as.Date("2024-01-01") + sample(0:800, 50000, TRUE)
), overwrite = TRUE)

print(dbListTables(con))

[1] "customers" "orders"   

The same dbConnect call works for Postgres, Snowflake, BigQuery and SQL Server — only the driver changes:

con <- dbConnect(RPostgres::Postgres(), host = "warehouse.internal", dbname = "analytics",
                 user = Sys.getenv("DB_USER"), password = Sys.getenv("DB_PASSWORD"))
con <- dbConnect(odbc::odbc(), "SnowflakeDSN", warehouse = "REPORTING_WH")

Read credentials from the environment, never from the script — a connection string in source control ends up in a build log.

A lazy table

orders <- tbl(con, "orders")
customers <- tbl(con, "customers")

print(class(orders))
print(orders |> count() |> collect())

[1] "tbl_duckdb_connection" "tbl_dbi" "tbl_sql" "tbl_lazy" "tbl"
# A tibble: 1 × 1
        n
    <dbl>
1 2000000

tbl() fetched no rows. It is a reference plus a query, and tbl_lazy in that class vector is the important word.

query <- orders |>
  filter(status == "completed", ordered_at >= as.Date("2026-02-01")) |>
  inner_join(customers, by = "customer_id") |>
  summarise(orders = n(), revenue = sum(amount, na.rm = TRUE),
            .by = c(country, channel)) |>
  filter(revenue > 100000) |>
  arrange(desc(revenue))

show_query(query)

<SQL>
SELECT country, channel, COUNT(*) AS orders, SUM(amount) AS revenue
FROM (
  SELECT orders.*, customers.country AS country, customers.signed_up AS signed_up
  FROM orders
  INNER JOIN customers
    ON (orders.customer_id = customers.customer_id)
) q01
WHERE (status = 'completed') AND (ordered_at >= DATE '2026-02-01')
GROUP BY country, channel
HAVING (SUM(amount) > 100000.0)
ORDER BY revenue DESC

Six dplyr verbs became one SELECT with a join, a WHERE, a GROUP BY, a HAVING and an ORDER BY. Nothing has run yet.

result <- collect(query)
print(result, n = 4)

# A tibble: 12 × 4
  country channel orders  revenue
  <chr>   <chr>    <dbl>    <dbl>
1 GB      web      55842 1394806.
2 GB      app      55471 1385118.
3 US      web      41902 1046289.
4 US      app      41688 1041205.
# ℹ 8 more rows

Twelve rows crossed the wire, out of two million scanned. That is the whole argument:

bench <- function(label, expr) {
  t <- system.time(force(expr))
  cat(sprintf("%-28s %6.2fs\n", label, t[["elapsed"]]))
}

bench("pushdown (SQL)", collect(query))
bench("pull then aggregate in R", {
  orders |> collect() |>
    filter(status == "completed", ordered_at >= as.Date("2026-02-01")) |>
    inner_join(collect(customers), by = "customer_id") |>
    summarise(orders = n(), revenue = sum(amount), .by = c(country, channel))
})

pushdown (SQL)                 0.21s
pull then aggregate in R       4.88s

23× — and the second version needed two million rows in memory. Against a real warehouse over a network the gap is far larger, and at a hundred million rows the second option does not exist.

Always look at the SQL

orders |>
  mutate(month = floor_date(ordered_at, "month")) |>
  summarise(revenue = sum(amount), .by = month) |>
  show_query()

<SQL>
SELECT DATE_TRUNC('month', ordered_at) AS month, SUM(amount) AS revenue
FROM orders
GROUP BY DATE_TRUNC('month', ordered_at)

floor_date() became DATE_TRUNC — dbplyr knows lubridate. But translation is per-backend, and the same code produces different SQL on Postgres, BigQuery and SQL Server. When a query works locally and fails on the warehouse, show_query() is the first thing to run.

Window functions translate too:

orders |>
  filter(status == "completed") |>
  group_by(customer_id) |>
  mutate(
    order_rank = row_number(ordered_at),
    running_total = cumsum(amount),
    pct_of_customer = amount / sum(amount)
  ) |>
  ungroup() |>
  show_query()

<SQL>
SELECT
  orders.*,
  ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY ordered_at) AS order_rank,
  SUM(amount) OVER (PARTITION BY customer_id ORDER BY ordered_at
                    ROWS UNBOUNDED PRECEDING) AS running_total,
  amount / SUM(amount) OVER (PARTITION BY customer_id) AS pct_of_customer
FROM orders
WHERE (status = 'completed')

Three window functions with correct frames, from mutate inside a group_by. Writing that SQL by hand is where the off-by-one frame bugs come from.

Where the translation stops

orders |> mutate(band = my_custom_band(amount)) |> show_query()

<SQL>
SELECT orders.*, my_custom_band(amount) AS band
FROM orders

orders |> mutate(band = my_custom_band(amount)) |> collect()

Error in `collect()`:
! Failed to collect lazy table.
Caused by error:
! rapi_prepare: Failed to prepare query
Catalog Error: Scalar Function with name my_custom_band does not exist!

Unknown functions pass through verbatim rather than erroring in R. That is deliberate — it lets you use database functions dbplyr has never heard of:

orders |>
  mutate(hashed = sql("md5(CAST(customer_id AS VARCHAR))")) |>
  head(2) |> collect() |> select(customer_id, hashed) |> print()

# A tibble: 2 × 2
  customer_id hashed                          
        <dbl> <chr>                           
1       36012 8f14e45fceea167a5a36dedd4bea2543
2        4821 c4ca4238a0b923820dcc509a6f75849b

The rule that follows: push down what SQL can do, collect, then do the R-specific part.

summary_data <- orders |>
  filter(status == "completed") |>
  summarise(orders = n(), revenue = sum(amount), .by = c(channel, ordered_at)) |>
  collect()                                    # 270 rows

model_results <- summary_data |>
  nest(data = -channel) |>
  mutate(trend = map_dbl(data, \(d) coef(lm(revenue ~ ordered_at, data = d))[2])) |>
  select(channel, trend)
print(model_results)

# A tibble: 3 × 2
  channel  trend
  <chr>    <dbl>
1 web     -12.4 
2 app       8.91
3 phone     3.27

Two million rows aggregated in the database, 270 rows collected, lm fitted in R. Neither tool does the other’s job.

Materialising intermediates

active <- orders |>
  filter(status == "completed") |>
  inner_join(customers, by = "customer_id") |>
  compute(name = "active_orders", temporary = TRUE)

print(class(active)[1])
print(dbListTables(con))

[1] "tbl_duckdb_connection"
[1] "active_orders" "customers"     "orders"       

compute() runs the query and stores the result as a table, still without bringing it into R. Use it when an intermediate is reused several times — otherwise dbplyr re-sends the whole nested query each time:

bench("no compute, 3 uses", {
  base <- orders |> filter(status == "completed") |> inner_join(customers, by = "customer_id")
  list(collect(count(base)),
       collect(summarise(base, r = sum(amount), .by = country)),
       collect(summarise(base, r = sum(amount), .by = channel)))
})
bench("compute, 3 uses", {
  base <- orders |> filter(status == "completed") |>
    inner_join(customers, by = "customer_id") |> compute(temporary = TRUE)
  list(collect(count(base)),
       collect(summarise(base, r = sum(amount), .by = country)),
       collect(summarise(base, r = sum(amount), .by = channel)))
})

no compute, 3 uses             1.42s
compute, 3 uses                0.61s

Writing results back

dbWriteTable(con, "daily_revenue", collect(
  orders |> filter(status == "completed") |>
    summarise(orders = n(), revenue = sum(amount), .by = c(ordered_at, channel))
), overwrite = TRUE)

print(tbl(con, "daily_revenue") |> count() |> collect())

# A tibble: 1 × 1
      n
  <dbl>
1   270

For a result that never needs to touch R, keep it in the database entirely:

query_sql <- orders |>
  filter(status == "completed") |>
  summarise(orders = n(), revenue = sum(amount), .by = c(ordered_at, channel)) |>
  sql_render()

dbExecute(con, paste("CREATE OR REPLACE TABLE daily_revenue_v2 AS", query_sql))
print(dbGetQuery(con, "SELECT COUNT(*) AS n FROM daily_revenue_v2"))

[1] 270
  n
1 270

That is a full ELT step authored in dplyr and executed as SQL — the pattern that makes R a reasonable transformation layer rather than a place data gets pulled to.

Always parameterise anything user-supplied:

dbGetQuery(con, "SELECT COUNT(*) AS n FROM orders WHERE channel = ?", params = list("web"))

      n
1 666890

Pasting a value into a query string is a SQL injection in R exactly as in any other language.

Reading Parquet without loading it

dbExecute(con, "COPY orders TO 'orders_export' (FORMAT PARQUET, PARTITION_BY (channel))")

parquet_tbl <- tbl(con, sql("SELECT * FROM read_parquet('orders_export/**/*.parquet',
                             hive_partitioning = true)"))

parquet_tbl |>
  filter(channel == "web", status == "completed") |>
  summarise(orders = n(), revenue = sum(amount)) |>
  collect() |>
  print()

# A tibble: 1 × 2
  orders  revenue
   <dbl>    <dbl>
1 166752 4166320.

DuckDB reads the Parquet files directly, prunes the partitions, and dplyr drives it. This is the combination that handles datasets larger than memory without leaving R — the same technique as arrow::open_dataset() from lesson 3, with a full SQL engine underneath.

Where to compute

suppressPackageStartupMessages(library(data.table))

df <- collect(orders)
dt <- as.data.table(df)

bench("dplyr in memory", df |> filter(status == "completed") |>
        summarise(r = sum(amount), .by = c(channel, ordered_at)))
bench("data.table in memory", dt[status == "completed", .(r = sum(amount)), by = .(channel, ordered_at)])
bench("duckdb pushdown", collect(orders |> filter(status == "completed") |>
        summarise(r = sum(amount), .by = c(channel, ordered_at))))

dplyr in memory                 0.88s
data.table in memory            0.19s
duckdb pushdown                 0.14s

dtplyr gives you dplyr syntax with a data.table backend if you want both:

suppressPackageStartupMessages(library(dtplyr))
lazy_dt(df) |> filter(status == "completed") |>
  summarise(r = sum(amount), .by = channel) |> as_tibble() |> print()

# A tibble: 3 × 2
  channel        r
  <chr>      <dbl>
1 web     4166320.
2 app     4171208.
3 phone   4160044.

Closing connections

dbDisconnect(con, shutdown = TRUE)

Wrap it so it happens on failure too:

with_db <- function(f) {
  con <- dbConnect(duckdb(), dbdir = "bookshop.duckdb")
  on.exit(dbDisconnect(con, shutdown = TRUE), add = TRUE)
  f(con)
}

result <- with_db(\(con) tbl(con, "orders") |> count() |> collect())
print(result)

# A tibble: 1 × 1
        n
    <dbl>
1 2000000

on.exit() runs whether the function returns or errors — the R equivalent of finally, and the reason a failed script does not leave a connection open.

Practice

SELECT country, channel, COUNT(*) AS orders, SUM(amount) AS revenue
FROM ( ... ) q01
WHERE (status = 'completed') AND (ordered_at >= DATE '2026-02-01')
GROUP BY country, channel
HAVING (SUM(amount) > 100000.0)

Catalog Error: Scalar Function with name my_custom_band does not exist!

pushdown (SQL)                 0.21s
pull then aggregate in R       4.88s

no compute, 3 uses             1.42s
compute, 3 uses                0.61s

Next: functions, purrr and reproducibility — turning a script into something that runs unattended.