From c6ff913358c7bfdda30107a3a9f08d4490e98438 Mon Sep 17 00:00:00 2001 From: Simon Couch Date: Mon, 10 Aug 2026 14:38:25 -0500 Subject: [PATCH 1/3] add databricks catalog discovery --- R/catalog-databricks.R | 413 ++++++++++++++++++++ R/data-source.R | 41 +- man/data_source.Rd | 7 +- tests/testthat/README.md | 8 +- tests/testthat/_snaps/catalog-databricks.md | 17 + tests/testthat/test-catalog-databricks.R | 90 +++++ tests/testthat/test-live-warehouses.R | 77 +++- 7 files changed, 634 insertions(+), 19 deletions(-) create mode 100644 R/catalog-databricks.R create mode 100644 tests/testthat/_snaps/catalog-databricks.md create mode 100644 tests/testthat/test-catalog-databricks.R diff --git a/R/catalog-databricks.R b/R/catalog-databricks.R new file mode 100644 index 0000000..f7a5bea --- /dev/null +++ b/R/catalog-databricks.R @@ -0,0 +1,413 @@ +is_databricks_connection <- function(con) { + info <- tryCatch(DBI::dbGetInfo(con), error = function(err) NULL) + labels <- c(info$drivername, info$sourcename) + labels <- labels[!is.na(labels)] + length(labels) > 0L && any(grepl("databricks", labels, ignore.case = TRUE)) +} + +databricks_table_registry <- function( + con, + tables = NULL, + call = rlang::caller_env() +) { + if (is.null(tables)) { + ids <- list(databricks_current_namespace(con, call = call)) + } else { + entries <- table_entries(tables, call = call) + ids <- lapply(entries, table_entry_id, call = call) + } + + relations <- list() + validate <- list() + for (id in ids) { + type <- databricks_id_type(id, call = call) + if (identical(type, "relation")) { + relations[[length(relations) + 1L]] <- databricks_exact_relation( + con, + id, + call = call + ) + validate[[length(validate) + 1L]] <- id + next + } + relations <- c( + relations, + databricks_list_relations(con, id, call = call) + ) + } + + labels <- vapply(relations, function(x) { + table_id_label(x$id, call = call) + }, character(1)) + duplicated_labels <- unique(labels[duplicated(labels)]) + if (length(duplicated_labels)) { + cli::cli_abort( + "{.arg tables} must not select duplicate labels: {.val {duplicated_labels}}.", + call = call + ) + } + + relation_ids <- lapply(relations, `[[`, "id") + names(relation_ids) <- labels + names(relations) <- labels + + validate_labels <- vapply(validate, table_id_label, character(1), call = call) + names(validate) <- validate_labels + + list( + labels = labels, + ids = relation_ids, + relations = relations, + validate = list(labels = validate_labels, ids = validate) + ) +} + +databricks_current_namespace <- function(con, call = rlang::caller_env()) { + row <- tryCatch( + DBI::dbGetQuery( + con, + paste( + "SELECT CURRENT_CATALOG() AS catalog,", + "CURRENT_SCHEMA() AS schema" + ) + ), + error = function(err) { + cli::cli_abort( + "Failed to read the current Databricks namespace.", + parent = err, + call = call + ) + } + ) + names(row) <- tolower(names(row)) + if (nrow(row) != 1L || !all(c("catalog", "schema") %in% names(row))) { + cli::cli_abort( + "Databricks returned an invalid current namespace.", + call = call + ) + } + values <- unlist(row[1, c("catalog", "schema")], use.names = TRUE) + if ( + length(values) != 2L || + any(is.na(values)) || + any(!nzchar(values)) + ) { + cli::cli_abort( + c( + "The Databricks connection has no current catalog and schema.", + "i" = "Set both on the connection or supply {.arg tables} as a {.cls DBI::Id}." + ), + call = call + ) + } + DBI::Id(catalog = values[["catalog"]], schema = values[["schema"]]) +} + +databricks_list_relations <- function( + con, + namespace, + call = rlang::caller_env() +) { + namespace <- databricks_complete_namespace(con, namespace, call = call) + components <- namespace@name + # `system.information_schema` does not expose legacy hive_metastore tables. + if (identical(tolower(components[["catalog"]]), "hive_metastore")) { + return(databricks_list_hive_relations(con, namespace, call = call)) + } + + databricks_list_unity_relations(con, namespace, call = call) +} + +databricks_list_unity_relations <- function( + con, + prefix, + call = rlang::caller_env() +) { + components <- prefix@name + + predicates <- c( + paste0( + "table_catalog = ", + DBI::dbQuoteString(con, components[["catalog"]]) + ), + "table_schema <> 'information_schema'", + if ("schema" %in% names(components)) { + paste0( + "table_schema = ", + DBI::dbQuoteString(con, components[["schema"]]) + ) + }, + if ("table" %in% names(components)) { + paste0( + "table_name = ", + DBI::dbQuoteString(con, components[["table"]]) + ) + } + ) + rows <- tryCatch( + DBI::dbGetQuery( + con, + paste( + "SELECT table_catalog, table_schema, table_name, table_type, comment", + "FROM system.information_schema.tables WHERE", + paste(predicates, collapse = " AND "), + "ORDER BY table_catalog, table_schema, table_name" + ) + ), + error = function(err) { + cli::cli_abort( + "Failed to list relations in the selected Databricks namespace.", + parent = err, + call = call + ) + } + ) + databricks_relations_from_information_schema(rows) +} + +databricks_exact_relation <- function(con, id, call = rlang::caller_env()) { + complete <- databricks_complete_relation(con, id, call = call) + components <- complete@name + namespace <- DBI::Id( + catalog = components[["catalog"]], + schema = components[["schema"]] + ) + relations <- if ( + identical(tolower(components[["catalog"]]), "hive_metastore") + ) { + databricks_list_hive_relations(con, namespace, call = call) + } else { + databricks_list_unity_relations(con, complete, call = call) + } + requested_name <- components[["table"]] + is_requested <- vapply( + relations, + function(relation) { + identical(relation$id@name[["table"]], requested_name) + }, + logical(1) + ) + if (!any(is_requested)) { + return(list(id = id, kind = NULL, description = NULL)) + } + + relation <- relations[[which(is_requested)[[1]]]] + relation$id <- id + relation +} + +databricks_relations_from_information_schema <- function(rows) { + names(rows) <- tolower(names(rows)) + lapply(seq_len(nrow(rows)), function(i) { + description <- rows$comment[[i]] + if (is.na(description) || !nzchar(description)) { + description <- NULL + } + kind <- if (grepl("view", rows$table_type[[i]], ignore.case = TRUE)) { + "view" + } else { + "table" + } + list( + id = DBI::Id( + catalog = rows$table_catalog[[i]], + schema = rows$table_schema[[i]], + table = rows$table_name[[i]] + ), + kind = kind, + description = description + ) + }) +} + +databricks_list_hive_relations <- function( + con, + namespace, + call = rlang::caller_env() +) { + components <- namespace@name + if (!"schema" %in% names(components)) { + cli::cli_abort( + "Selecting {.val hive_metastore} requires a schema-level {.cls DBI::Id}.", + call = call + ) + } + rows <- tryCatch( + DBI::dbGetQuery( + con, + paste("SHOW TABLES IN", DBI::dbQuoteIdentifier(con, namespace)) + ), + error = function(err) { + cli::cli_abort( + "Failed to list relations in the selected hive_metastore schema.", + parent = err, + call = call + ) + } + ) + names(rows) <- tolower(names(rows)) + lapply(rows$tablename, function(table) { + list( + id = DBI::Id( + catalog = components[["catalog"]], + schema = components[["schema"]], + table = table + ), + kind = NULL, + description = NULL + ) + }) +} + +databricks_describe_relation <- function(con, id, call = rlang::caller_env()) { + rows <- tryCatch( + DBI::dbGetQuery( + con, + paste("DESCRIBE TABLE", DBI::dbQuoteIdentifier(con, id)) + ), + error = function(err) { + cli::cli_abort( + "Failed to describe the selected Databricks relation.", + parent = err, + call = call + ) + } + ) + complete <- databricks_complete_relation(con, id, call = call) + nullable <- databricks_column_nullability(con, complete, call = call) + databricks_columns_from_describe(rows, nullable) +} + +databricks_column_nullability <- function( + con, + id, + call = rlang::caller_env() +) { + components <- id@name + if (identical(tolower(components[["catalog"]]), "hive_metastore")) { + return(logical()) + } + predicates <- c( + paste0("table_catalog = ", DBI::dbQuoteString(con, components[["catalog"]])), + paste0("table_schema = ", DBI::dbQuoteString(con, components[["schema"]])), + paste0("table_name = ", DBI::dbQuoteString(con, components[["table"]])) + ) + rows <- tryCatch( + DBI::dbGetQuery( + con, + paste( + "SELECT column_name, is_nullable", + "FROM system.information_schema.columns WHERE", + paste(predicates, collapse = " AND ") + ) + ), + error = function(err) { + cli::cli_abort( + "Failed to read Databricks column nullability.", + parent = err, + call = call + ) + } + ) + names(rows) <- tolower(names(rows)) + nullable <- rep(NA, nrow(rows)) + is_nullable <- toupper(rows$is_nullable) + nullable[is_nullable %in% "YES"] <- TRUE + nullable[is_nullable %in% "NO"] <- FALSE + stats::setNames(nullable, rows$column_name) +} + +databricks_columns_from_describe <- function(rows, nullable = logical()) { + names(rows) <- tolower(names(rows)) + metadata <- which(startsWith(rows$col_name, "#")) + if (length(metadata)) { + rows <- rows[seq_len(metadata[[1]] - 1L), , drop = FALSE] + } + rows <- rows[ + !is.na(rows$col_name) & + nzchar(rows$col_name), + , + drop = FALSE + ] + description <- rows$comment + description[is.na(description) | !nzchar(description)] <- NA_character_ + column_nullable <- unname(nullable[rows$col_name]) + if (length(nullable) == 0L) { + column_nullable <- rep(NA, nrow(rows)) + } + data.frame( + column = rows$col_name, + type = rows$data_type, + nullable = column_nullable, + description = description, + row.names = NULL + ) +} + +databricks_complete_namespace <- function( + con, + id, + call = rlang::caller_env() +) { + components <- id@name + if (!"catalog" %in% names(components)) { + current <- databricks_current_namespace(con, call = call)@name + catalog <- current[["catalog"]] + } else { + catalog <- components[["catalog"]] + } + if ("schema" %in% names(components)) { + return(DBI::Id(catalog = catalog, schema = components[["schema"]])) + } + DBI::Id(catalog = catalog) +} + +databricks_complete_relation <- function( + con, + id, + call = rlang::caller_env() +) { + components <- id@name + if (!all(c("catalog", "schema") %in% names(components))) { + current <- databricks_current_namespace(con, call = call)@name + } else { + current <- NULL + } + DBI::Id( + catalog = if ("catalog" %in% names(components)) { + components[["catalog"]] + } else { + current[["catalog"]] + }, + schema = if ("schema" %in% names(components)) { + components[["schema"]] + } else { + current[["schema"]] + }, + table = components[["table"]] + ) +} + +databricks_id_type <- function(id, call = rlang::caller_env()) { + components <- id@name + roles <- names(components) + valid <- list( + c("catalog"), + c("schema"), + c("catalog", "schema"), + c("table"), + c("schema", "table"), + c("catalog", "schema", "table") + ) + if ( + !any(vapply(valid, identical, logical(1), roles)) || + any(is.na(components) | !nzchar(components)) + ) { + cli::cli_abort( + "Databricks {.cls DBI::Id} entries in {.arg tables} must follow + catalog, schema, and table order without skipped or empty components.", + call = call + ) + } + if ("table" %in% roles) "relation" else "namespace" +} diff --git a/R/data-source.R b/R/data-source.R index 4e6cfc3..b59e9cd 100644 --- a/R/data-source.R +++ b/R/data-source.R @@ -35,9 +35,10 @@ #' strings like `"schema.table"`, or `DBI::Id` objects. Defaults to every #' table returned by [DBI::dbListTables()]. Strings containing dots are #' interpreted as schema-qualified names; use `DBI::Id(table = "a.b")` for -#' literal table names containing dots. For Snowflake connections, a -#' `DBI::Id` ending in `catalog` or `schema` selects every table and view in -#' that namespace. Leaving `tables` unset selects the current Snowflake +#' literal table names containing dots. For Snowflake and Databricks +#' connections, a `DBI::Id` ending in `catalog` or `schema` selects every +#' table and view in that namespace. Leaving `tables` unset selects the +#' current schema. A Databricks `hive_metastore` selection must include a #' schema. #' #' For a board, a named character vector of pins to read: the names become @@ -162,7 +163,27 @@ data_source_connection <- function( owned = FALSE, table_ids = table_registry$ids, dictionary = dictionary, - snowflake_relations = table_registry$relations + relations = table_registry$relations + )) + } + + if (is_databricks_connection(con)) { + table_registry <- databricks_table_registry(con, tables, call = call) + if (length(table_registry$validate$labels)) { + check_table_ids_exist(con, table_registry$validate, call = call) + } + commons_span_set_attribute( + span, + "commons.data_source.n_tables", + length(table_registry$labels) + ) + return(new_data_source( + con, + table_registry$labels, + owned = FALSE, + table_ids = table_registry$ids, + dictionary = dictionary, + relations = table_registry$relations )) } @@ -265,7 +286,7 @@ new_data_source <- function( table_ids = table_ids_from_labels(tables), dictionary = NULL, pending = NULL, - snowflake_relations = NULL + relations = NULL ) { # Disconnect only the DuckDB connection we created; a user-supplied connection # has its own owner and lifetime. @@ -288,7 +309,7 @@ new_data_source <- function( handle = handle, dictionary = dictionary, pending = pending, - snowflake_relations = snowflake_relations + relations = relations ), class = "commons_data_source" ) @@ -474,15 +495,17 @@ source_describe <- function( n_sample ) ) - relation <- source$snowflake_relations[[table]] - if (is.null(source$snowflake_relations)) { + relation <- source$relations[[table]] + if (is.null(source$relations)) { schema <- data.frame( column = names(sample), type = vapply(sample, function(x) class(x)[[1]], character(1)), row.names = NULL ) - } else { + } else if (is_snowflake_connection(source$con)) { schema <- snowflake_describe_relation(source$con, id, call = call) + } else { + schema <- databricks_describe_relation(source$con, id, call = call) } list( schema = schema, diff --git a/man/data_source.Rd b/man/data_source.Rd index 9f8c240..f7dcb52 100644 --- a/man/data_source.Rd +++ b/man/data_source.Rd @@ -18,9 +18,10 @@ For a connection, a character vector of table names, schema-qualified strings like \code{"schema.table"}, or \code{DBI::Id} objects. Defaults to every table returned by \code{\link[DBI:dbListTables]{DBI::dbListTables()}}. Strings containing dots are interpreted as schema-qualified names; use \code{DBI::Id(table = "a.b")} for -literal table names containing dots. For Snowflake connections, a -\code{DBI::Id} ending in \code{catalog} or \code{schema} selects every table and view in -that namespace. Leaving \code{tables} unset selects the current Snowflake +literal table names containing dots. For Snowflake and Databricks +connections, a \code{DBI::Id} ending in \code{catalog} or \code{schema} selects every +table and view in that namespace. Leaving \code{tables} unset selects the +current schema. A Databricks \code{hive_metastore} selection must include a schema. For a board, a named character vector of pins to read: the names become diff --git a/tests/testthat/README.md b/tests/testthat/README.md index e31877c..e3a70dd 100644 --- a/tests/testthat/README.md +++ b/tests/testthat/README.md @@ -25,8 +25,8 @@ options(commons.test.databricks = DBI::Id( )) ``` -The Snowflake test exercises exact table selection, schema and catalog +Both warehouse tests exercise exact table selection, schema and catalog expansion, current-schema discovery, native column metadata, and sample rows. -The Databricks test queries the current identity and namespace, then reads at -most one row from the configured table. If a backend's option is absent, its -test skips. Once enabled, connection and query failures fail the test. +The Databricks test also uses a temporary view to exercise quoted relation and +column names. If a backend's option is absent, its test skips. Once enabled, +connection and query failures fail the test. diff --git a/tests/testthat/_snaps/catalog-databricks.md b/tests/testthat/_snaps/catalog-databricks.md new file mode 100644 index 0000000..cf6afa4 --- /dev/null +++ b/tests/testthat/_snaps/catalog-databricks.md @@ -0,0 +1,17 @@ +# Databricks identifiers cannot skip components + + Code + databricks_id_type(DBI::Id(catalog = "main", table = "orders")) + Condition + Error: + ! Databricks entries in `tables` must follow catalog, schema, and table order without skipped or empty components. + +# Databricks current namespace requires a catalog and schema + + Code + databricks_current_namespace(NULL) + Condition + Error: + ! The Databricks connection has no current catalog and schema. + i Set both on the connection or supply `tables` as a . + diff --git a/tests/testthat/test-catalog-databricks.R b/tests/testthat/test-catalog-databricks.R new file mode 100644 index 0000000..374aa54 --- /dev/null +++ b/tests/testthat/test-catalog-databricks.R @@ -0,0 +1,90 @@ +test_that("Databricks information schema retains native relation metadata", { + rows <- data.frame( + table_catalog = c("Data.Catalog", "main"), + table_schema = c("Odd Schema", "default"), + table_name = c("Sales.Report", "orders"), + table_type = c("VIEW", "MANAGED"), + comment = c("A useful view", ""), + stringsAsFactors = FALSE + ) + + relations <- databricks_relations_from_information_schema(rows) + + expect_length(relations, 2) + expect_equal(relations[[1]]$kind, "view") + expect_equal(relations[[1]]$description, "A useful view") + expect_equal(relations[[2]]$kind, "table") + expect_null(relations[[2]]$description) + expect_identical( + relations[[1]]$id, + DBI::Id( + catalog = "Data.Catalog", + schema = "Odd Schema", + table = "Sales.Report" + ) + ) + + con <- DBI::dbConnect(duckdb::duckdb()) + withr::defer(DBI::dbDisconnect(con, shutdown = TRUE)) + expect_equal( + as.character(DBI::dbQuoteIdentifier(con, relations[[1]]$id)), + '"Data.Catalog"."Odd Schema"."Sales.Report"' + ) +}) + +test_that("Databricks descriptions retain columns and nullability", { + rows <- data.frame( + col_name = c( + "Order ID", + "notes", + "# Partition Information", + "notes" + ), + data_type = c("bigint", "string", "", "string"), + comment = c("Primary key", NA, "", NA), + stringsAsFactors = FALSE + ) + nullable <- c("Order ID" = FALSE, notes = TRUE) + + columns <- databricks_columns_from_describe(rows, nullable) + + expect_equal(columns$column, c("Order ID", "notes")) + expect_equal(columns$type, c("bigint", "string")) + expect_identical(columns$nullable, c(FALSE, TRUE)) + expect_equal(columns$description, c("Primary key", NA)) +}) + +test_that("Databricks identifiers distinguish namespaces and relations", { + expect_equal(databricks_id_type(DBI::Id(catalog = "main")), "namespace") + expect_equal(databricks_id_type(DBI::Id(schema = "default")), "namespace") + expect_equal( + databricks_id_type(DBI::Id(catalog = "main", schema = "default")), + "namespace" + ) + expect_equal(databricks_id_type(DBI::Id(table = "orders")), "relation") + expect_equal( + databricks_id_type(DBI::Id(schema = "default", table = "orders")), + "relation" + ) +}) + +test_that("Databricks identifiers cannot skip components", { + expect_snapshot( + databricks_id_type(DBI::Id(catalog = "main", table = "orders")), + error = TRUE + ) +}) + +test_that("Databricks current namespace requires a catalog and schema", { + local_mocked_bindings( + dbGetQuery = function(...) { + data.frame(catalog = NA_character_, schema = NA_character_) + }, + .package = "DBI" + ) + + expect_snapshot( + databricks_current_namespace(NULL), + error = TRUE + ) +}) diff --git a/tests/testthat/test-live-warehouses.R b/tests/testthat/test-live-warehouses.R index 9510b64..c9faf3d 100644 --- a/tests/testthat/test-live-warehouses.R +++ b/tests/testthat/test-live-warehouses.R @@ -62,12 +62,12 @@ test_that("live Snowflake discovers and describes catalog relations", { ) expect_true(nrow(described$sample) <= 5) expect_equal(names(described$sample), described$schema$column) - expect_true(exact$snowflake_relations[[label]]$kind %in% c("table", "view")) + expect_true(exact$relations[[label]]$kind %in% c("table", "view")) expect_true(label %in% list_tables(schema_source)) expect_true(label %in% list_tables(catalog_source)) expect_true(label %in% list_tables(current_source)) expect_true(all(vapply( - schema_source$snowflake_relations, + schema_source$relations, function(x) x$kind %in% c("table", "view"), logical(1) ))) @@ -78,9 +78,14 @@ test_that("live Snowflake discovers and describes catalog relations", { expect_match(tool@value, "Sample summary") }) -test_that("live Databricks connection reads a configured table", { +test_that("live Databricks discovers and describes catalog relations", { table <- warehouse_test_table("databricks") con <- local_warehouse_connection("databricks") + components <- table@name + skip_if_not( + all(c("catalog", "schema", "table") %in% names(components)), + "The Databricks test table must be fully qualified" + ) session <- DBI::dbGetQuery( con, @@ -92,10 +97,76 @@ test_that("live Databricks connection reads a configured table", { ) rows <- warehouse_read_one(con, table) names(session) <- tolower(names(session)) + label <- table_id_label(table) + + exact <- data_source(con, tables = table) + described <- source_describe(exact, label) + + namespace <- DBI::Id( + catalog = components[["catalog"]], + schema = components[["schema"]] + ) + schema_source <- data_source(con, tables = namespace) + catalog_source <- data_source( + con, + tables = DBI::Id(catalog = components[["catalog"]]) + ) + current_source <- data_source(con) expect_equal(nrow(session), 1) expect_named(session, c("principal", "catalog", "schema")) expect_true(nzchar(session$principal[[1]])) expect_s3_class(rows, "data.frame") expect_true(nrow(rows) <= 1) + expect_equal(list_tables(exact), label) + expect_identical(exact$table_ids[[label]], table) + expect_named( + described$schema, + c("column", "type", "nullable", "description") + ) + expect_false(anyNA(described$schema$nullable)) + expect_true(nrow(described$sample) <= 5) + expect_equal(names(described$sample), described$schema$column) + expect_true(exact$relations[[label]]$kind %in% c("table", "view")) + expect_true(label %in% list_tables(schema_source)) + expect_true(label %in% list_tables(catalog_source)) + expect_s3_class(current_source, "commons_data_source") + expect_true(all(vapply( + schema_source$relations, + function(x) x$kind %in% c("table", "view"), + logical(1) + ))) + + tool <- describe_table_tool(exact, label) + expect_match(tool@value, "Relation type") + expect_match(tool@value, "nullable") + expect_match(tool@value, "Sample summary") +}) + +test_that("live Databricks handles quoted relation and column names", { + warehouse_test_table("databricks") + con <- local_warehouse_connection("databricks") + DBI::dbExecute(con, "USE CATALOG `hive_metastore`") + DBI::dbExecute(con, "USE SCHEMA `default`") + + table <- DBI::Id(table = "commons quoted.table") + quoted <- DBI::dbQuoteIdentifier(con, table) + DBI::dbExecute( + con, + paste( + "CREATE TEMPORARY VIEW", + quoted, + "AS SELECT 1 AS `quoted column`" + ) + ) + withr::defer( + DBI::dbExecute(con, paste("DROP VIEW IF EXISTS", quoted)) + ) + + source <- data_source(con, tables = table) + described <- source_describe(source, "commons quoted.table") + + expect_identical(source$table_ids[["commons quoted.table"]], table) + expect_equal(described$schema$column, "quoted column") + expect_equal(described$sample[["quoted column"]], 1L) }) From fbfadfdf039e3fcf34eaf341a904a13b80c2d8b8 Mon Sep 17 00:00:00 2001 From: Simon Couch Date: Mon, 10 Aug 2026 14:48:12 -0500 Subject: [PATCH 2/3] share catalog registry assembly --- R/catalog-databricks.R | 57 ++++++------------------------------------ R/catalog-snowflake.R | 55 ++++++---------------------------------- R/catalog.R | 54 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 70 insertions(+), 96 deletions(-) create mode 100644 R/catalog.R diff --git a/R/catalog-databricks.R b/R/catalog-databricks.R index f7a5bea..d4c198a 100644 --- a/R/catalog-databricks.R +++ b/R/catalog-databricks.R @@ -10,55 +10,14 @@ databricks_table_registry <- function( tables = NULL, call = rlang::caller_env() ) { - if (is.null(tables)) { - ids <- list(databricks_current_namespace(con, call = call)) - } else { - entries <- table_entries(tables, call = call) - ids <- lapply(entries, table_entry_id, call = call) - } - - relations <- list() - validate <- list() - for (id in ids) { - type <- databricks_id_type(id, call = call) - if (identical(type, "relation")) { - relations[[length(relations) + 1L]] <- databricks_exact_relation( - con, - id, - call = call - ) - validate[[length(validate) + 1L]] <- id - next - } - relations <- c( - relations, - databricks_list_relations(con, id, call = call) - ) - } - - labels <- vapply(relations, function(x) { - table_id_label(x$id, call = call) - }, character(1)) - duplicated_labels <- unique(labels[duplicated(labels)]) - if (length(duplicated_labels)) { - cli::cli_abort( - "{.arg tables} must not select duplicate labels: {.val {duplicated_labels}}.", - call = call - ) - } - - relation_ids <- lapply(relations, `[[`, "id") - names(relation_ids) <- labels - names(relations) <- labels - - validate_labels <- vapply(validate, table_id_label, character(1), call = call) - names(validate) <- validate_labels - - list( - labels = labels, - ids = relation_ids, - relations = relations, - validate = list(labels = validate_labels, ids = validate) + catalog_table_registry( + con, + tables, + current_namespace = databricks_current_namespace, + id_type = databricks_id_type, + exact_relation = databricks_exact_relation, + list_relations = databricks_list_relations, + call = call ) } diff --git a/R/catalog-snowflake.R b/R/catalog-snowflake.R index 09a9faf..8fba1ea 100644 --- a/R/catalog-snowflake.R +++ b/R/catalog-snowflake.R @@ -9,53 +9,14 @@ snowflake_table_registry <- function( tables = NULL, call = rlang::caller_env() ) { - if (is.null(tables)) { - ids <- list(snowflake_current_namespace(con, call = call)) - } else { - entries <- table_entries(tables, call = call) - ids <- lapply(entries, table_entry_id, call = call) - } - - relations <- list() - validate <- list() - for (id in ids) { - type <- snowflake_id_type(id, call = call) - if (identical(type, "relation")) { - relation <- snowflake_exact_relation(con, id, call = call) - label <- table_id_label(id, call = call) - relations[[length(relations) + 1L]] <- relation - validate[[length(validate) + 1L]] <- id - next - } - relations <- c( - relations, - snowflake_list_relations(con, id, call = call) - ) - } - - labels <- vapply(relations, function(x) { - table_id_label(x$id, call = call) - }, character(1)) - duplicated_labels <- unique(labels[duplicated(labels)]) - if (length(duplicated_labels)) { - cli::cli_abort( - "{.arg tables} must not select duplicate labels: {.val {duplicated_labels}}.", - call = call - ) - } - - relation_ids <- lapply(relations, `[[`, "id") - names(relation_ids) <- labels - names(relations) <- labels - - validate_labels <- vapply(validate, table_id_label, character(1), call = call) - names(validate) <- validate_labels - - list( - labels = labels, - ids = relation_ids, - relations = relations, - validate = list(labels = validate_labels, ids = validate) + catalog_table_registry( + con, + tables, + current_namespace = snowflake_current_namespace, + id_type = snowflake_id_type, + exact_relation = snowflake_exact_relation, + list_relations = snowflake_list_relations, + call = call ) } diff --git a/R/catalog.R b/R/catalog.R new file mode 100644 index 0000000..5fb476a --- /dev/null +++ b/R/catalog.R @@ -0,0 +1,54 @@ +catalog_table_registry <- function( + con, + tables, + current_namespace, + id_type, + exact_relation, + list_relations, + call = rlang::caller_env() +) { + if (is.null(tables)) { + ids <- list(current_namespace(con, call = call)) + } else { + entries <- table_entries(tables, call = call) + ids <- lapply(entries, table_entry_id, call = call) + } + + relations <- list() + validate <- list() + for (id in ids) { + type <- id_type(id, call = call) + if (identical(type, "relation")) { + relation <- exact_relation(con, id, call = call) + relations[[length(relations) + 1L]] <- relation + validate[[length(validate) + 1L]] <- id + next + } + relations <- c(relations, list_relations(con, id, call = call)) + } + + labels <- vapply(relations, function(x) { + table_id_label(x$id, call = call) + }, character(1)) + duplicated_labels <- unique(labels[duplicated(labels)]) + if (length(duplicated_labels)) { + cli::cli_abort( + "{.arg tables} must not select duplicate labels: {.val {duplicated_labels}}.", + call = call + ) + } + + relation_ids <- lapply(relations, `[[`, "id") + names(relation_ids) <- labels + names(relations) <- labels + + validate_labels <- vapply(validate, table_id_label, character(1), call = call) + names(validate) <- validate_labels + + list( + labels = labels, + ids = relation_ids, + relations = relations, + validate = list(labels = validate_labels, ids = validate) + ) +} From 81f0589f71ddf26fece7ce0fa432cec5a75161d4 Mon Sep 17 00:00:00 2001 From: Simon Couch Date: Mon, 10 Aug 2026 14:52:00 -0500 Subject: [PATCH 3/3] share catalog selection policy --- R/catalog-databricks.R | 38 ++------------------------------------ R/catalog-snowflake.R | 38 ++------------------------------------ R/catalog.R | 42 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 46 insertions(+), 72 deletions(-) diff --git a/R/catalog-databricks.R b/R/catalog-databricks.R index d4c198a..6e1965f 100644 --- a/R/catalog-databricks.R +++ b/R/catalog-databricks.R @@ -138,21 +138,7 @@ databricks_exact_relation <- function(con, id, call = rlang::caller_env()) { } else { databricks_list_unity_relations(con, complete, call = call) } - requested_name <- components[["table"]] - is_requested <- vapply( - relations, - function(relation) { - identical(relation$id@name[["table"]], requested_name) - }, - logical(1) - ) - if (!any(is_requested)) { - return(list(id = id, kind = NULL, description = NULL)) - } - - relation <- relations[[which(is_requested)[[1]]]] - relation$id <- id - relation + catalog_match_exact_relation(relations, id) } databricks_relations_from_information_schema <- function(rows) { @@ -348,25 +334,5 @@ databricks_complete_relation <- function( } databricks_id_type <- function(id, call = rlang::caller_env()) { - components <- id@name - roles <- names(components) - valid <- list( - c("catalog"), - c("schema"), - c("catalog", "schema"), - c("table"), - c("schema", "table"), - c("catalog", "schema", "table") - ) - if ( - !any(vapply(valid, identical, logical(1), roles)) || - any(is.na(components) | !nzchar(components)) - ) { - cli::cli_abort( - "Databricks {.cls DBI::Id} entries in {.arg tables} must follow - catalog, schema, and table order without skipped or empty components.", - call = call - ) - } - if ("table" %in% roles) "relation" else "namespace" + catalog_id_type(id, "Databricks", call = call) } diff --git a/R/catalog-snowflake.R b/R/catalog-snowflake.R index 8fba1ea..bc88915 100644 --- a/R/catalog-snowflake.R +++ b/R/catalog-snowflake.R @@ -107,21 +107,7 @@ snowflake_exact_relation <- function(con, id, call = rlang::caller_env()) { } ) relations <- snowflake_relations_from_show(rows) - requested_name <- components[["table"]] - is_requested <- vapply( - relations, - function(relation) { - identical(relation$id@name[["table"]], requested_name) - }, - logical(1) - ) - if (!any(is_requested)) { - return(list(id = id, kind = NULL, description = NULL)) - } - - relation <- relations[[which(is_requested)[[1]]]] - relation$id <- id - relation + catalog_match_exact_relation(relations, id) } snowflake_relations_from_show <- function(rows) { @@ -175,25 +161,5 @@ snowflake_describe_relation <- function(con, id, call = rlang::caller_env()) { } snowflake_id_type <- function(id, call = rlang::caller_env()) { - components <- id@name - roles <- names(components) - valid <- list( - c("catalog"), - c("schema"), - c("catalog", "schema"), - c("table"), - c("schema", "table"), - c("catalog", "schema", "table") - ) - if ( - !any(vapply(valid, identical, logical(1), roles)) || - any(is.na(components) | !nzchar(components)) - ) { - cli::cli_abort( - "Snowflake {.cls DBI::Id} entries in {.arg tables} must follow - catalog, schema, and table order without skipped or empty components.", - call = call - ) - } - if ("table" %in% roles) "relation" else "namespace" + catalog_id_type(id, "Snowflake", call = call) } diff --git a/R/catalog.R b/R/catalog.R index 5fb476a..f41b010 100644 --- a/R/catalog.R +++ b/R/catalog.R @@ -52,3 +52,45 @@ catalog_table_registry <- function( validate = list(labels = validate_labels, ids = validate) ) } + +catalog_id_type <- function(id, backend, call = rlang::caller_env()) { + components <- id@name + roles <- names(components) + valid <- list( + c("catalog"), + c("schema"), + c("catalog", "schema"), + c("table"), + c("schema", "table"), + c("catalog", "schema", "table") + ) + if ( + !any(vapply(valid, identical, logical(1), roles)) || + any(is.na(components) | !nzchar(components)) + ) { + cli::cli_abort( + "{backend} {.cls DBI::Id} entries in {.arg tables} must follow + catalog, schema, and table order without skipped or empty components.", + call = call + ) + } + if ("table" %in% roles) "relation" else "namespace" +} + +catalog_match_exact_relation <- function(relations, id) { + requested_name <- id@name[["table"]] + is_requested <- vapply( + relations, + function(relation) { + identical(relation$id@name[["table"]], requested_name) + }, + logical(1) + ) + if (!any(is_requested)) { + return(list(id = id, kind = NULL, description = NULL)) + } + + relation <- relations[[which(is_requested)[[1]]]] + relation$id <- id + relation +}