Code Walkthrough

This vignette explains how datapond works under the hood. It walks through each file in the package, explaining what the code does and why it’s designed that way.

Package Structure

datapond/
├── DESCRIPTION          # Package metadata
├── NAMESPACE            # Exports and imports
├── LICENSE              # MIT license
├── R/
│   ├── db_connect.R     # Connection management
│   ├── db_read.R        # Read functions
│   ├── db_write.R       # Write functions
│   ├── upsert.R         # MERGE operations
│   ├── metadata.R       # Snapshot/catalog queries
│   ├── discovery.R      # List/exists functions
│   ├── maintenance.R    # Vacuum, rollback, diff, compaction
│   ├── advanced.R       # Clustering, inlining, Iceberg export
│   ├── arrow.R          # Arrow integration (read/write)
│   ├── docs.R           # Documentation & data dictionary
│   ├── preview.R        # Write preview functions
│   ├── browser.R        # Shiny browser launcher
│   ├── browser_ui.R     # Shiny browser UI module
│   ├── browser_server.R # Shiny browser server module
│   └── zzz.R            # Package load/unload hooks
└── vignettes/
    ├── concepts.Rmd     # Conceptual background
    ├── catalog-backends.Rmd  # Catalog backend options
    └── code-walkthrough.Rmd  # This file

Core Design Decisions

1. Singleton Connection Pattern

The package maintains a single connection stored in a private environment:

.db_env <- new.env(parent = emptyenv())

Why?

Prevents users accidentally creating multiple connections
Simplifies the API (no need to pass connection objects around)
Ensures cleanup happens properly

All functions retrieve the connection via .db_get_con() rather than creating new ones.

2. Input Validation

All user inputs are validated early with clear error messages:

.db_validate_name <- function(x, arg = "name") {
  # Must be single non-empty string
  # Only allows A-Z a-z 0-9 _ -
  # Prevents SQL injection and path traversal
}

This catches problems like table = "../../../etc/passwd" before they cause harm.

3. Lazy Evaluation with DuckDB

Read functions return lazy dplyr tables, not collected data:

dplyr::tbl(con, dplyr::sql(query))

Why?

Large datasets aren’t loaded into memory until needed
DuckDB can optimise the full query (filters, joins, aggregations)
Users call collect() when they’re ready to bring data into R

4. DuckLake SQL Parameter Syntax

DuckLake uses two different syntaxes for named parameters depending on the function:

=> for DuckLake extension functions:

# Functions like ducklake_merge_adjacent_files, ducklake_list_files, etc.
CALL ducklake_merge_adjacent_files('catalog', 'table', schema => 'main')
FROM ducklake_list_files('catalog', 'table', schema => 'main')

:= for catalog-level set_option calls:

# Setting options like data_inlining_row_limit
CALL catalog.set_option('data_inlining_row_limit', 100, table_name := 'events')

This distinction matters because using the wrong syntax will cause SQL errors. The package handles this internally, but it’s important to understand when reading the generated SQL.

File-by-File Walkthrough

`R/db_connect.R` - Connection Management

This file handles connecting to and disconnecting from DuckLake.

Private Environment and Helpers

.db_env <- new.env(parent = emptyenv())

This creates an isolated environment that persists for the R session. We store:

con - the DuckDB connection object
data_path - root path for parquet files
catalog - DuckLake catalog name
catalog_type - “duckdb”, “sqlite”, or “postgres”
metadata_path - path to catalog file/connection string

.db_get_con <- function() {
  # Returns the connection if it exists and is valid, NULL otherwise
  if (exists("con", envir = .db_env) && DBI::dbIsValid(.db_env$con)) {
    return(.db_env$con)
  }
  NULL
}

This is the safe way to get the connection - it checks validity, not just existence.

.db_get <- function(name, default = NULL) {
  # Safe getter with default value
  if (exists(name, envir = .db_env)) {
    get(name, envir = .db_env)
  } else {
    default
  }
}

Used throughout the package to retrieve stored values.

Catalog Backend Helpers

These internal functions handle the different catalog backend types:

.db_build_ducklake_dsn <- function(catalog_type, metadata_path) {
  switch(catalog_type,
    duckdb = paste0("ducklake:", metadata_path),
    sqlite = paste0("ducklake:sqlite:", metadata_path),
    postgres = paste0("ducklake:postgres:", metadata_path),
    stop("Unknown catalog_type")
  )
}

This builds the correct DuckLake connection string based on the backend: - DuckDB: ducklake:metadata.ducklake - SQLite: ducklake:sqlite:catalog.sqlite - PostgreSQL: ducklake:postgres:dbname=... host=...

.db_load_catalog_extensions <- function(con, catalog_type) {
  # Always need ducklake
  try(DBI::dbExecute(con, "INSTALL ducklake"), silent = TRUE)
  DBI::dbExecute(con, "LOAD ducklake")

  # Load backend-specific extension
  if (catalog_type == "sqlite") {
    try(DBI::dbExecute(con, "INSTALL sqlite"), silent = TRUE)
    DBI::dbExecute(con, "LOAD sqlite")
  } else if (catalog_type == "postgres") {
    try(DBI::dbExecute(con, "INSTALL postgres"), silent = TRUE)
    DBI::dbExecute(con, "LOAD postgres")
  }
}

DuckDB needs extra extensions loaded for SQLite and PostgreSQL backends.

`db_connect()` - DuckLake Connection

db_connect <- function(duckdb_db = ":memory:",
                       catalog = "cso",
                       catalog_type = c("duckdb", "sqlite", "postgres"),
                       metadata_path = "metadata.ducklake",
                       data_path = "/data/lake",
                       snapshot_version = NULL,
                       snapshot_time = NULL,
                       ...) {

  catalog_type <- match.arg(catalog_type)

  # Create DuckDB connection
  con <- DBI::dbConnect(duckdb::duckdb(), dbdir = duckdb_db)

  # Load DuckLake and backend-specific extensions
  .db_load_catalog_extensions(con, catalog_type)

  # Build the backend-specific connection string
  ducklake_dsn <- .db_build_ducklake_dsn(catalog_type, metadata_path)

  # Build ATTACH statement with options
  attach_opts <- c(glue::glue("DATA_PATH {.db_sql_quote(data_path)}"))

  if (!is.null(snapshot_version)) {
    attach_opts <- c(attach_opts, glue::glue("SNAPSHOT_VERSION {as.integer(snapshot_version)}"))
  }

  attach_sql <- glue::glue(
    "ATTACH {.db_sql_quote(ducklake_dsn)} AS {catalog} ({paste(attach_opts, collapse = ', ')})"
  )

  # Attach with helpful error messages
  tryCatch({
    DBI::dbExecute(con, attach_sql)
  }, error = function(e) {
    DBI::dbDisconnect(con, shutdown = TRUE)
    hint <- switch(catalog_type,
      sqlite = "Ensure the sqlite extension is available and the metadata file path is accessible.",
      postgres = "Ensure PostgreSQL is running and the connection string is correct.",
      duckdb = "Ensure the metadata file path is accessible."
    )
    stop("Failed to attach DuckLake catalog.\n", hint, "\n\nOriginal error: ", e$message)
  })

  DBI::dbExecute(con, glue::glue("USE {catalog}"))

  # Store connection info
  assign("con", con, envir = .db_env)
  assign("catalog", catalog, envir = .db_env)
  assign("catalog_type", catalog_type, envir = .db_env)
  assign("metadata_path", metadata_path, envir = .db_env)
  assign("data_path", data_path, envir = .db_env)
  ...
}

Key points:

Supports three catalog backends (DuckDB, SQLite, PostgreSQL)
ATTACH statement connects the DuckLake catalog to the DuckDB session
Error messages include backend-specific troubleshooting hints

`R/db_write.R` - Writing Data

`db_write()` - DuckLake Tables

db_write() supports two input types:

data.frame/tibble - Data passes through R memory
Lazy dbplyr table - Data stays entirely in DuckDB (zero-copy)

db_write <- function(data, schema = "main", table,
                     mode = c("overwrite", "append"),
                     col_types = NULL,       # Explicit column types (data.frame only)
                     partition_by = NULL,    # Hive partitioning
                     bucket_by = NULL,       # Bucket partitioning
                     sort_by = NULL,         # Clustering
                     inline = FALSE,         # Data inlining
                     commit_author = NULL,
                     commit_message = NULL) {

  # Detect if input is a lazy dbplyr table
  is_lazy <- inherits(data, "tbl_lazy") || inherits(data, "tbl_sql")

  if (is_lazy) {
    # ===== LAZY TABLE PATH =====
    # Extract SQL from the dbplyr query
    select_sql <- dbplyr::sql_render(data)

    # Execute directly in DuckDB - no R memory used
    if (mode == "overwrite") {
      DBI::dbExecute(con, glue("CREATE TABLE {qname} AS {select_sql}"))
    } else {
      DBI::dbExecute(con, glue("INSERT INTO {qname} {select_sql}"))
    }

  } else {
    # ===== DATA.FRAME PATH =====
    # Register data temporarily in DuckDB
    tmp <- .db_temp_name()
    duckdb::duckdb_register(con, tmp, data)

    # Build column definitions with explicit types
    cols <- vapply(data, .db_r_to_duckdb_type, character(1))
    if (!is.null(col_types)) {
      for (col_name in names(col_types)) {
        cols[col_name] <- col_types[[col_name]]
      }
    }

    # Create table and insert from temp view
    DBI::dbExecute(con, glue("CREATE TABLE {qname} ({col_defs})"))
    DBI::dbExecute(con, glue("INSERT INTO {qname} SELECT * FROM {tmp}"))
  }

  # Both paths: apply partitioning, clustering, etc.
}

Key points:

Uses transactions (BEGIN/COMMIT) for atomicity
ROLLBACK on error ensures no partial changes
Lazy tables use dbplyr::sql_render() to extract SQL, then CREATE TABLE AS SELECT
col_types only applies to data.frame input (lazy tables get types from query)
Supports hive partitioning, bucket partitioning, clustering, and data inlining

Example: Zero-copy pipeline

# Data never leaves DuckDB - efficient for in-lake transforms
db_read(table = "raw_events") |>
  filter(year == 2024) |>
  group_by(event_type) |>
  summarise(count = n()) |>
  db_write(table = "event_counts")
# Message: Wrote data to cso.main.event_counts (from query)

`R/arrow.R` - Arrow Integration

# Read directly as Arrow Table (bypasses DuckDB compute)
db_read_arrow <- function(schema = "main", table, columns = NULL, as_data_frame = TRUE) {
  # Get parquet file paths from DuckLake metadata
  files <- .db_get_table_files(con, catalog, schema, table)

  # Read via Arrow
  if (length(files) == 1) {
    arrow::read_parquet(files[1], col_select = columns)
  } else {
    arrow::open_dataset(files) |> dplyr::collect()
  }
}

# Write Arrow Table to DuckLake
db_write_arrow <- function(data, schema = "main", table, mode = c("overwrite", "append"), ...) {
  # Convert Arrow to data.frame and delegate to db_write
  if (inherits(data, c("ArrowTabular", "Table"))) {
    data <- as.data.frame(data)
  }
  db_write(data, schema, table, mode, ...)
}

When to use Arrow functions:

db_read_arrow(): Direct Parquet access, Arrow ecosystem interop, large data transfers
db_write_arrow(): When data is already in Arrow format

`R/upsert.R` - MERGE Operations

`db_upsert()` - Update + Insert

db_upsert <- function(data, schema = "main", table, by,
                      strict = TRUE, update_cols = NULL,
                      commit_author = NULL, commit_message = NULL) {

  # Validate key columns exist in both data and target
  missing_keys <- setdiff(by, names(data))
  if (length(missing_keys) > 0) stop(...)

  # Strict mode: reject duplicate keys in incoming data
  if (strict) {
    dup_check_sql <- glue::glue("
      SELECT {by_sql}, COUNT(*) AS n
      FROM {tmp}
      GROUP BY {by_sql}
      HAVING COUNT(*) > 1
    ")
    dups <- DBI::dbGetQuery(con, dup_check_sql)
    if (nrow(dups) > 0) stop("Duplicate keys found...")
  }

  # Build MERGE statement
  on_sql <- paste(glue::glue("t.{by} = s.{by}"), collapse = " AND ")

  # UPDATE clause depends on update_cols
  update_clause <- if (is.null(update_cols)) {
    "WHEN MATCHED THEN UPDATE"  # DuckDB shorthand: update all
  } else if (length(update_cols) == 0) {
    ""  # Insert-only mode
  } else {
    glue::glue("WHEN MATCHED THEN UPDATE SET {update_sql}")
  }

  merge_sql <- glue::glue("
    MERGE INTO {qname} AS t
    USING {tmp} AS s
    ON ({on_sql})
    {update_clause}
    WHEN NOT MATCHED THEN INSERT ...
  ")

  # Execute in transaction
  DBI::dbExecute(con, "BEGIN")
  DBI::dbExecute(con, merge_sql)
  DBI::dbExecute(con, "COMMIT")
}

Key points:

MERGE INTO is standard SQL for upsert operations
strict = TRUE prevents accidental duplicates (a common data quality issue)
update_cols = NULL means update everything; character(0) means insert-only
All wrapped in a transaction for safety

`R/docs.R` - Documentation & Data Dictionary

Metadata Storage

Documentation metadata is stored using DuckLake’s native COMMENT ON with JSON for structured data. This approach:

Survives disconnect/reconnect (unlike custom metadata tables)
Supports time travel (comments are versioned with snapshots)
Is stored in the DuckLake catalog alongside table definitions

# Internal helper to parse JSON comment
.db_parse_comment <- function(comment_str) {
  if (is.null(comment_str) || !nzchar(comment_str)) return(NULL)

  tryCatch({
    parsed <- jsonlite::fromJSON(comment_str)
    if (is.list(parsed)) return(parsed)
    list(description = comment_str)  # Plain string
  }, error = function(e) {
    list(description = comment_str)  # Not JSON, treat as description
  })
}

# Get table comment from DuckDB catalog
.db_get_table_comment <- function(con, catalog, schema, table) {
  result <- DBI::dbGetQuery(con, glue::glue("
    SELECT comment FROM duckdb_tables()
    WHERE database_name = ... AND schema_name = ... AND table_name = ...
  "))
  if (nrow(result) == 0 || is.na(result$comment[1])) return(NULL)
  result$comment[1]
}

`db_get_docs()` - Retrieve Documentation

db_get_docs <- function(schema = "main", table) {
  # Read table comment from duckdb_tables()
  table_comment <- .db_get_table_comment(con, catalog, schema, table)
  parsed <- .db_parse_comment(table_comment)

  # Build result with defaults

  result <- list(
    schema = schema, table = table,
    description = parsed$description,
    owner = parsed$owner,
    tags = parsed$tags %||% character(0),
    lineage = if (!is.null(parsed$lineage_sources)) list(
      sources = parsed$lineage_sources,
      transformation = parsed$lineage_transformation
    ),
    columns = list()  # Populated from duckdb_columns()
  )

  # Get column comments
  cols <- DBI::dbGetQuery(con, "SELECT column_name, comment FROM duckdb_columns() ...")
  for (col in cols with comments) {
    result$columns[[col_name]] <- .db_parse_comment(col_comment)
  }

  result
}

`db_dictionary()` - Generate Data Dictionary

db_dictionary <- function(schema = NULL, include_columns = TRUE) {
  # Query information_schema for tables
  tables <- DBI::dbGetQuery(con, "SELECT * FROM information_schema.tables ...")

  for (tbl in tables) {
    # Get docs from native comments via db_get_docs()
    docs <- db_get_docs(schema = ..., table = ...)
    cols <- DBI::dbGetQuery(con, "SELECT * FROM information_schema.columns ...")
  }

  do.call(rbind, rows)
}

`db_search()` - Find Tables

db_search <- function(pattern, field = c("all", "name", "description", "owner", "tags")) {
  # Get full dictionary (without columns for speed)
  dict <- db_dictionary(include_columns = FALSE)

  # Filter by pattern match
  matches <- switch(field,
    all = grepl(pattern, dict$name) | grepl(pattern, dict$description) | ...,
    name = grepl(pattern, dict$name),
    description = grepl(pattern, dict$description),
    ...
  )

  dict[matches, ]
}

`db_lineage()` - Record Data Lineage

Design: static analysis via dplyneage

db_lineage() supports two modes. In the simple mode, the caller supplies sources and an optional transformation string. In the richer mode, the caller passes the actual dbplyr pipeline that produced the table, and the optional dplyneage package analyses its query tree to extract column-level lineage:

db_lineage <- function(schema = "main", table, sources = NULL,
                       transformation = NULL, pipeline = NULL) {

  # pipeline must be a lazy dbplyr table (not collected data)
  if (!is.null(pipeline)) {
    if (!inherits(pipeline, c("tbl_lazy", "tbl_sql"))) {
      stop("'pipeline' must be a lazy dbplyr table (tbl_lazy/tbl_sql).")
    }

    # dplyneage is optional - graceful degradation when absent
    if (!requireNamespace("dplyneage", quietly = TRUE)) {
      message("Install dplyneage for automatic column-level lineage: ",
              "pak::pak('tgerke/dplyneage')")
    } else {
      # Extract lineage by walking the dbplyr query tree - no SQL execution
      lin   <- dplyneage::extract_lineage(pipeline)
      edges <- dplyneage::lineage_edges(lin)
      col_edges <- as.list(edges)

      # Infer table-level sources from the edge data when not provided
      if (is.null(sources)) {
        sources <- unique(edges$source_table)
        sources <- sources[nzchar(sources)]
      }
    }
  }
  ...
}

dplyneage::extract_lineage() inspects the dbplyr query object (its lazy expression tree) rather than executing it. It traces each output column back through joins, aggregations, and computed expressions to the originating source column. This means no data is read from the database during lineage extraction - it is purely static analysis of the dplyr call graph.

Storage: lineage_column_edges alongside other metadata

Column edges are serialised as a list and stored in the table’s JSON comment under the key lineage_column_edges, alongside the existing description, owner, tags, and lineage_sources fields:

# The comment JSON for a table with full metadata looks like:
{
  "description": "Aggregated import totals by country",
  "owner": "Trade Section",
  "tags": ["trade", "monthly"],
  "lineage_sources": ["main.imports"],
  "lineage_transformation": "Grouped by country, summed value",
  "lineage_column_edges": [
    { "source_table": "main.imports", "source_column": "value",
      "target_column": "total", "transformation": "SUM" },
    { "source_table": "main.imports", "source_column": "country",
      "target_column": "country", "transformation": "identity" }
  ]
}

Because this sits inside the native COMMENT ON TABLE statement, it:

Survives disconnect/reconnect (no separate metadata table needed)
Is versioned alongside the table in DuckLake snapshots (time-travel aware)
Merges cleanly with any existing description, owner, or tags (the merge reads the current comment, updates only the lineage keys, and writes the whole object back)

# Merge logic - never overwrites description/owner/tags
existing_meta <- .db_parse_comment(.db_get_table_comment(...)) %||% list()

if (!is.null(sources))       existing_meta$lineage_sources        <- sources
if (!is.null(transformation)) existing_meta$lineage_transformation <- transformation
if (!is.null(col_edges))     existing_meta$lineage_column_edges   <- col_edges

db_comment(schema = schema, table = table, comment = existing_meta)

Graceful degradation when dplyneage is absent

requireNamespace("dplyneage", quietly = TRUE) is the standard R idiom for a soft optional dependency. When dplyneage is not installed:

A single informative message() is printed (not a warning, not an error)
The function continues and records table-level lineage only (sources + transformation string), if those were provided
db_write(track_lineage = TRUE) silently falls back to no lineage rather than aborting a write

This means the package works fully without dplyneage; the extra package only unlocks column-level detail.

`db_get_lineage()` - Retrieve lineage

db_get_lineage <- function(schema = "main", table) {
  parsed <- .db_parse_comment(.db_get_table_comment(con, catalog, schema, table))

  result <- list(
    sources        = parsed$lineage_sources,
    transformation = parsed$lineage_transformation
  )

  # column_edges is only present when dplyneage was used
  if (!is.null(parsed$lineage_column_edges)) {
    result$column_edges <- as.data.frame(parsed$lineage_column_edges)
  }

  result
}

column_edges is returned as a proper data.frame with one row per source-to-target column relationship. When only table-level lineage was recorded (no pipeline= argument, or dplyneage was absent), column_edges is simply absent from the returned list.

`db_lineage_flow()` - Visualise column lineage

db_lineage_flow <- function(schema = "main", table) {
  lin <- db_get_lineage(schema = schema, table = table)

  # Reconstruct a dplyneage lineage object from the stored edge data.frame
  dplyneage::lineage_flow(
    dplyneage::lineage_from_edges(lin$column_edges)
  )
}

The edges stored in JSON are sufficient to reconstruct the dplyneage lineage object via lineage_from_edges(). lineage_flow() then renders an interactive React Flow diagram in the viewer. This means the original pipeline object does not need to be kept around - the serialised edges capture everything needed for rendering.

`db_write(track_lineage = TRUE)` - One-step shortcut

For the common case of writing a dbplyr pipeline and immediately wanting lineage recorded, db_write() accepts track_lineage = TRUE:

db_read(table = "imports") |>
  dplyr::group_by(country) |>
  dplyr::summarise(total = sum(value)) |>
  db_write(table = "country_totals", track_lineage = TRUE)
# Wrote data to cso.main.country_totals (from query)
# Recorded column-level lineage for cso.main.country_totals

Internally, db_write() calls db_lineage(schema, table, pipeline = data) after a successful write, passing the same lazy table object it used for the write:

if (isTRUE(track_lineage)) {
  if (!is_lazy) {
    message("track_lineage ignored: lineage tracking requires a lazy dbplyr pipeline.")
  } else {
    tryCatch(
      db_lineage(schema = schema, table = table, pipeline = data),
      error = function(e) message("Lineage tracking failed: ", e$message)
    )
  }
}

track_lineage is only meaningful for lazy table input. For data.frame input, a message is printed and the write continues normally. Lineage failures are caught with tryCatch so a lineage extraction error never aborts a completed write.

`R/preview.R` - Write Previews

`db_preview_write()` - Preview Before Writing

db_preview_write <- function(data, schema = "main", table,
                             mode = c("overwrite", "append")) {

  preview <- list(
    mode = mode,
    table_exists = db_table_exists(schema = schema, table = table),
    incoming = list(
      rows = nrow(data),
      cols = ncol(data),
      columns = names(data)
    )
  )

  # If target exists, compare schemas
  if (preview$table_exists) {
    existing_cols <- db_table_cols(schema = schema, table = table)

    preview$schema_changes <- list(
      new_columns = setdiff(names(data), existing_cols),
      removed_columns = setdiff(existing_cols, names(data))
    )
  }

  .print_preview(preview)
  invisible(preview)
}

Key points:

Gathers all relevant information without making any changes
Compares incoming schema with existing schema
Returns structured data for programmatic use, prints human-readable summary

`db_preview_upsert()` - Preview Insert vs Update Counts

db_preview_upsert <- function(data, schema, table, by, update_cols = NULL) {

  # Register data temporarily
  tmp <- .db_temp_name()
  duckdb::duckdb_register(con, tmp, data)

  # Count matches (updates)
  match_sql <- glue::glue("
    SELECT COUNT(*) FROM {tmp} s
    INNER JOIN {qname} t ON {join_on_keys}
  ")
  matches <- DBI::dbGetQuery(con, match_sql)$n

  # Check for duplicate keys
  dup_sql <- glue::glue("
    SELECT {key_cols}, COUNT(*) FROM {tmp}
    GROUP BY {key_cols} HAVING COUNT(*) > 1
  ")
  dups <- DBI::dbGetQuery(con, dup_sql)

  preview$impact <- list(
    inserts = nrow(data) - matches,
    updates = matches,
    duplicates_in_incoming = nrow(dups)
  )

  .print_upsert_preview(preview)
  invisible(preview)
}

`R/discovery.R` - Finding Data

DuckLake Discovery

db_tables <- function(schema = "main") {
  sql <- glue::glue("
    SELECT table_name
    FROM information_schema.tables
    WHERE table_catalog = {.db_sql_quote(catalog)}
      AND table_schema  = {.db_sql_quote(schema)}
      AND table_type    = 'BASE TABLE'
  ")
  DBI::dbGetQuery(con, sql)$table_name
}

Key points:

Uses standard information_schema views
Filters by catalog to avoid confusion with other attached databases

Views and Macros

Views and macros are stored in the DuckLake catalog and support time travel:

# Create a view - stored in catalog
db_create_view(view = "active_users",
               query = "SELECT * FROM users WHERE active = true")

# Create a scalar macro
db_create_macro(name = "add_vals", params = c("a", "b"), body = "a + b")

# Create a table macro (returns a table, not a scalar)
db_create_macro(name = "big_sales",
                params = c(threshold = "INTEGER"),
                body = "SELECT * FROM sales WHERE value > threshold",
                table_macro = TRUE)

# List and drop
db_list_views()
db_list_macros()
db_drop_view(view = "active_users")
db_drop_macro(name = "add_vals")

Documentation with db_comment()

db_comment() accepts either a list (stored as JSON) or a string (stored as-is):

# Structured metadata (stored as JSON)
db_comment(table = "users", comment = list(
  description = "Active user accounts",
  owner = "Admin Team",
  tags = c("users", "reference")
))

# Simple string comment
db_comment(table = "users", comment = "Active user accounts")

# Column documentation
db_comment(table = "users", column = "email", comment = list(
  description = "Primary email",
  units = NULL
))

# Remove comment
db_comment(table = "users", column = "email", comment = NULL)

The list format is automatically converted to JSON via jsonlite::toJSON(). When retrieved via db_get_docs(), JSON is parsed back to a list.

Logging

Enable DuckDB query logging for debugging:

db_enable_logging(TRUE, log_type = "query")
# ... run operations ...
db_query("SELECT * FROM duckdb_logs()")
db_enable_logging(FALSE)

`R/maintenance.R` - Admin Operations

`db_vacuum()` - Clean Up Old Snapshots

db_vacuum <- function(older_than = "30 days", dry_run = TRUE) {
  # Get snapshots before
  snapshots_before <- db_snapshots()

  if (dry_run) {
    # Calculate what would be removed
    # Show preview
    return(invisible(to_remove))
  }

  # Actually vacuum
  DBI::dbExecute(con, glue::glue(
    "CALL ducklake_vacuum({.db_sql_quote(catalog)}, INTERVAL '30 days')"
  ))
}

Key points:

dry_run = TRUE by default prevents accidental deletion
Shows clear before/after summary

`db_rollback()` - Restore Previous Version

db_rollback <- function(schema, table, version = NULL, timestamp = NULL) {
  # Build time travel clause
  at_clause <- glue::glue("AT (VERSION => {version})")

  # Rollback = create new version with old data
  rollback_sql <- glue::glue(
    "CREATE OR REPLACE TABLE {qname} AS SELECT * FROM {qname} {at_clause}"
  )

  DBI::dbExecute(con, rollback_sql)
}

Key points:

Rollback creates a NEW snapshot with the old data (non-destructive)
You can always “roll forward” by rolling back to a later version

`db_diff()` - Compare Versions

db_diff <- function(schema, table, from_version, to_version, key_cols = NULL) {
  # ADDED: rows in 'to' but not in 'from'
  added_sql <- glue::glue("SELECT * FROM {to_ref} EXCEPT SELECT * FROM {from_ref}")

  # REMOVED: rows in 'from' but not in 'to'
  removed_sql <- glue::glue("SELECT * FROM {from_ref} EXCEPT SELECT * FROM {to_ref}")

  result <- list(
    added = DBI::dbGetQuery(con, added_sql),
    removed = DBI::dbGetQuery(con, removed_sql)
  )

  # If key_cols provided, find modified rows (same key, different values)
  if (!is.null(key_cols)) {
    # Modified = key appears in both added and removed
    modified_sql <- glue::glue("
      SELECT new_tbl.*
      FROM ({added_sql}) AS new_tbl
      INNER JOIN ({removed_sql}) AS old_tbl
      ON {join_on_keys}
    ")
    result$modified <- DBI::dbGetQuery(con, modified_sql)
  }

  result
}

`db_compact()` - Merge Small Files

db_compact <- function(schema = "main", table = NULL, max_files = NULL) {
  # Build the SQL call with optional parameters
  sql <- glue::glue("CALL ducklake_merge_adjacent_files({catalog}, {table}, ...)")

  # Get file stats before and after for reporting
  stats_before <- db_file_stats(schema, table)
  DBI::dbExecute(con, sql)
  stats_after <- db_file_stats(schema, table)

  # Report reduction
  message("Files before: ", sum(stats_before$file_count))
  message("Files after:  ", sum(stats_after$file_count))
}

Key points:

Uses DuckLake’s ducklake_merge_adjacent_files() procedure
max_files parameter limits memory usage for large tables
Files with incompatible schema versions cannot be merged
Run db_cleanup_files() after to remove old files

`db_file_stats()` - Check File Statistics

db_file_stats <- function(schema = NULL, table = NULL) {
  # Query DuckLake metadata tables directly
  # Tables are in __ducklake_metadata_{catalog} schema
  meta_schema <- glue::glue("__ducklake_metadata_{catalog}")

  sql <- glue::glue("
    SELECT
      s.schema_name, t.table_name,
      COUNT(f.data_file_id) AS file_count,
      COALESCE(SUM(f.record_count), 0) AS total_rows,
      COALESCE(SUM(f.file_size_bytes), 0) AS total_bytes
    FROM {meta_schema}.ducklake_table t
    JOIN {meta_schema}.ducklake_schema s ON t.schema_id = s.schema_id
    LEFT JOIN {meta_schema}.ducklake_data_file f
      ON t.table_id = f.table_id AND f.end_snapshot IS NULL
    WHERE t.end_snapshot IS NULL AND s.end_snapshot IS NULL
    GROUP BY s.schema_name, t.table_name
  ")
  info <- DBI::dbGetQuery(con, sql)

  info$avg_file_bytes <- ifelse(info$file_count > 0, info$total_bytes / info$file_count, 0)
  info$avg_rows_per_file <- ifelse(info$file_count > 0, info$total_rows / info$file_count, 0)
  info
}

Key points:

Queries DuckLake metadata tables directly (ducklake_table, ducklake_schema, ducklake_data_file)
Filters by end_snapshot IS NULL to get only current (not deleted) data
High file count with small average size indicates compaction needed
Useful for monitoring and maintenance scheduling

`db_cleanup_files()` - Remove Orphaned Files

db_cleanup_files <- function(dry_run = TRUE) {
  if (dry_run) {
    # Show preview of what would be cleaned
    return(invisible(NA))
  }

  sql <- glue::glue("CALL ducklake_cleanup_old_files({.db_sql_quote(catalog)})")
  DBI::dbExecute(con, sql)
}

Key points:

Orphaned files are created by db_vacuum() and db_compact()
dry_run = TRUE by default for safety
Reclaims disk space by removing unreferenced Parquet files

`R/advanced.R` - Advanced Features

This file contains advanced DuckLake features: clustering, inlining, and Iceberg export.

`db_set_clustering()` - Configure Sort Order

db_set_clustering <- function(schema = "main", table, columns) {
  qname <- paste0(catalog, ".", schema, ".", table)

  if (is.null(columns) || length(columns) == 0) {
    sql <- glue::glue("ALTER TABLE {qname} RESET SORTED BY")
  } else {
    sort_clause <- paste(columns, collapse = ", ")
    sql <- glue::glue("ALTER TABLE {qname} SET SORTED BY ({sort_clause})")
  }

  DBI::dbExecute(con, sql)
}

Key points:

Uses DuckLake’s ALTER TABLE ... SET SORTED BY to configure sort order
Clustering improves performance for range scans on sorted columns
Only affects new data - use db_recluster() for existing data

`db_recluster()` - Re-sort Existing Data

db_recluster <- function(schema = "main", table, max_files = NULL) {
  # DuckLake sorts data during compaction based on SET SORTED BY order
  # Use ducklake_merge_adjacent_files to compact and re-sort
  args <- c(.db_sql_quote(catalog), .db_sql_quote(table))
  named_args <- c(glue::glue("schema => {.db_sql_quote(schema)}"))

  if (!is.null(max_files)) {
    named_args <- c(named_args, glue::glue("max_compacted_files => {as.integer(max_files)}"))
  }

  all_args <- c(args, named_args)
  sql <- glue::glue("CALL ducklake_merge_adjacent_files({paste(all_args, collapse = ', ')})")

  DBI::dbExecute(con, sql)
}

Key points:

Uses ducklake_merge_adjacent_files() to compact and re-sort data
DuckLake automatically applies the table’s sort order during compaction
Note the => syntax for named parameters (standard for DuckLake functions)

`db_set_inline_threshold()` - Configure Data Inlining

db_set_inline_threshold <- function(schema = "main", table = NULL, threshold = 10) {
  # Build set_option call using := syntax for named parameters
  args <- c("'data_inlining_row_limit'", as.integer(threshold))

  if (!is.null(table)) {
    args <- c(args, glue::glue("table_name := {.db_sql_quote(table)}"))
  }
  if (!is.null(schema)) {
    args <- c(args, glue::glue("schema := {.db_sql_quote(schema)}"))
  }

  sql <- glue::glue("CALL {catalog}.set_option({paste(args, collapse = ', ')})")
  DBI::dbExecute(con, sql)
}

Key points:

Uses catalog’s set_option() procedure to configure inlining threshold
Note the := syntax for named parameters (specific to set_option)
Inlining stores small writes in the catalog instead of creating parquet files
Flush with db_flush_inlined() when ready to write to parquet

`R/zzz.R` - Package Hooks

# Null coalesce operator used throughout
`%||%` <- function(x, y) if (is.null(x) || length(x) == 0) y else x

.onAttach <- function(libname, pkgname) {
  packageStartupMessage(
    "datapond ", utils::packageVersion("datapond"), "\n",
    "Use db_connect() to connect to a DuckLake catalog."
  )
}

.onUnload <- function(libpath) {
  # Clean up connection when package unloads
  con <- .db_get_con()
  if (!is.null(con)) {
    try(DBI::dbDisconnect(con, shutdown = TRUE), silent = TRUE)
  }
}

Key points:

.onAttach runs when the package is loaded (shows helpful message)
.onUnload ensures cleanup if the package is unloaded
%||% handles both NULL and zero-length values

`R/browser.R`, `R/browser_ui.R`, `R/browser_server.R` - Interactive Browser

These files implement a Shiny gadget for interactively browsing the data lake.

Architecture

The browser uses the Shiny module pattern, which allows it to be: 1. Used standalone via db_browser() 2. Embedded in other Shiny apps via db_browser_ui() + db_browser_server()

# Standalone usage
db_browser()

# Embedded in another app
ui <- fluidPage(
  db_browser_ui("my_browser")
)
server <- function(input, output, session) {
  db_browser_server("my_browser")
}

`browser.R` - Launcher

db_browser <- function(height = "500px", viewer = c("dialog", "browser", "pane")) {
  # Check packages are available
  .db_assert_browser_packages()

  # Build app
  ui <- db_browser_app_ui(height = height)
  server <- function(input, output, session) {
    db_browser_server(id = "db_browser", height = height)
  }
  app <- shiny::shinyApp(ui = ui, server = server)

  # Run as gadget
  shiny::runGadget(app, viewer = viewer_func, stopOnCancel = TRUE)
}

`browser_ui.R` - Module UI

The UI is built with bslib for modern Bootstrap 5 styling:

db_browser_ui <- function(id, height = "500px") {
  ns <- shiny::NS(id)  # Namespace for module

  # Sidebar with tree view
  sidebar <- bslib::sidebar(
    shiny::uiOutput(ns("tree_view")),
    shiny::actionButton(ns("refresh_tree"), "Refresh")
  )

  # Main content with tabs
  main_content <- bslib::navset_card_tab(
    bslib::nav_panel("Preview", ...),
    bslib::nav_panel("Metadata", ...),
    bslib::nav_panel("Search", ...),
    bslib::nav_panel("Dictionary", ...)
  )

  bslib::page_sidebar(sidebar = sidebar, main_content)
}

Key points:

NS(id) creates namespaced IDs so multiple instances don’t conflict
bslib provides modern Bootstrap 5 components
Tree view is rendered dynamically based on available schemas and tables

`browser_server.R` - Module Server

db_browser_server <- function(id, height = "500px") {
  shiny::moduleServer(id, function(input, output, session) {

    # Reactive values for selection state
    rv <- shiny::reactiveValues(
      selected_schema = NULL,
      selected_table = NULL
    )

    # Tree view
    output$tree_view <- shiny::renderUI({
      .render_ducklake_tree(ns, rv)
    })

    # Preview - loads data on button click
    preview_data <- shiny::eventReactive(input$load_preview, {
      db_read(rv$selected_schema, rv$selected_table) |>
        head(input$preview_rows) |>
        collect()
    })

    output$preview_table <- DT::renderDataTable({
      DT::datatable(preview_data(), filter = "top")
    })

    # Search
    search_results <- shiny::eventReactive(input$do_search, {
      db_search(input$search_pattern, field = input$search_field)
    })

    # Dictionary with download
    output$download_dict <- shiny::downloadHandler(
      filename = "data_dictionary.csv",
      content = function(file) {
        write.csv(db_dictionary(), file)
      }
    )
  })
}

Key points:

moduleServer() creates the namespaced server logic
reactiveValues tracks UI state (selected table, etc.)
eventReactive triggers expensive operations only on button click
DT::datatable provides interactive tables with filtering

Security Considerations

SQL Injection Prevention

All user inputs that go into SQL are either:

Validated with .db_validate_name() (only allows A-Za-z0-9_-)
Quoted with .db_sql_quote() (escapes single quotes)

# This is safe:
table <- .db_validate_name(table)  # Rejects "../../../etc"
path <- .db_sql_quote(glob_path)   # Escapes quotes properly

Path Traversal Prevention

.db_validate_name() rejects any input containing: - Path separators (/, \) - Parent directory references (..) - Special characters

This prevents attacks like:

db_read(table = "../../../etc/passwd")  # Rejected!

Testing Your Understanding

Try answering these questions:

Why do we use .db_get_con() instead of accessing .db_env$con directly?
What happens if db_write() fails halfway through?
Why does db_read() return a lazy table instead of collected data?
What’s the difference between update_cols = NULL and update_cols = character(0) in db_upsert()?
Why would you choose SQLite over DuckDB as a catalog backend?
How does db_preview_upsert() know how many rows will be updated vs inserted?
Where is documentation metadata stored? (Hint: native COMMENT ON with JSON)

Contributing to the Package

When adding new functions:

Always validate inputs with .db_validate_name() or custom validation
Check connection at the start of every function
Use .db_get() and .db_get_con() for accessing stored state
Return invisibly for side-effect functions, visibly for queries
Add roxygen documentation with @param, @return, @examples, @export
Update NAMESPACE if adding new exports
Add tests in tests/testthat/ for new functionality

Run devtools::document() after adding roxygen comments to regenerate NAMESPACE and help files.

Package Structure

Core Design Decisions

1. Singleton Connection Pattern

2. Input Validation

3. Lazy Evaluation with DuckDB

4. DuckLake SQL Parameter Syntax

File-by-File Walkthrough

R/db_connect.R - Connection Management

Private Environment and Helpers

Catalog Backend Helpers

db_connect() - DuckLake Connection

R/db_write.R - Writing Data

db_write() - DuckLake Tables

R/arrow.R - Arrow Integration

R/upsert.R - MERGE Operations

db_upsert() - Update + Insert

R/docs.R - Documentation & Data Dictionary

Metadata Storage

db_get_docs() - Retrieve Documentation

db_dictionary() - Generate Data Dictionary

db_search() - Find Tables

db_lineage() - Record Data Lineage

Design: static analysis via dplyneage

Storage: lineage_column_edges alongside other metadata

Graceful degradation when dplyneage is absent

db_get_lineage() - Retrieve lineage

db_lineage_flow() - Visualise column lineage

db_write(track_lineage = TRUE) - One-step shortcut

R/preview.R - Write Previews

db_preview_write() - Preview Before Writing

db_preview_upsert() - Preview Insert vs Update Counts

R/discovery.R - Finding Data

DuckLake Discovery

Views and Macros

Documentation with db_comment()

Logging

R/maintenance.R - Admin Operations

db_vacuum() - Clean Up Old Snapshots

db_rollback() - Restore Previous Version

db_diff() - Compare Versions

db_compact() - Merge Small Files

db_file_stats() - Check File Statistics

db_cleanup_files() - Remove Orphaned Files

R/advanced.R - Advanced Features

db_set_clustering() - Configure Sort Order

db_recluster() - Re-sort Existing Data

db_set_inline_threshold() - Configure Data Inlining

R/zzz.R - Package Hooks

R/browser.R, R/browser_ui.R, R/browser_server.R - Interactive Browser

Architecture

browser.R - Launcher

browser_ui.R - Module UI

browser_server.R - Module Server

Security Considerations

SQL Injection Prevention

Path Traversal Prevention

Testing Your Understanding

Contributing to the Package

`R/db_connect.R` - Connection Management

`db_connect()` - DuckLake Connection

`R/db_write.R` - Writing Data

`db_write()` - DuckLake Tables

`R/arrow.R` - Arrow Integration

`R/upsert.R` - MERGE Operations

`db_upsert()` - Update + Insert

`R/docs.R` - Documentation & Data Dictionary

`db_get_docs()` - Retrieve Documentation

`db_dictionary()` - Generate Data Dictionary

`db_search()` - Find Tables

`db_lineage()` - Record Data Lineage

`db_get_lineage()` - Retrieve lineage

`db_lineage_flow()` - Visualise column lineage

`db_write(track_lineage = TRUE)` - One-step shortcut

`R/preview.R` - Write Previews

`db_preview_write()` - Preview Before Writing

`db_preview_upsert()` - Preview Insert vs Update Counts

`R/discovery.R` - Finding Data

`R/maintenance.R` - Admin Operations

`db_vacuum()` - Clean Up Old Snapshots

`db_rollback()` - Restore Previous Version

`db_diff()` - Compare Versions

`db_compact()` - Merge Small Files

`db_file_stats()` - Check File Statistics

`db_cleanup_files()` - Remove Orphaned Files

`R/advanced.R` - Advanced Features

`db_set_clustering()` - Configure Sort Order

`db_recluster()` - Re-sort Existing Data

`db_set_inline_threshold()` - Configure Data Inlining

`R/zzz.R` - Package Hooks

`R/browser.R`, `R/browser_ui.R`, `R/browser_server.R` - Interactive Browser

`browser.R` - Launcher

`browser_ui.R` - Module UI

`browser_server.R` - Module Server