add snowflake catalog discovery

R/catalog-snowflake.R

@@ -0,0 +1,238 @@
+is_snowflake_connection <- function(con) {
+  info <- tryCatch(DBI::dbGetInfo(con), error = function(err) NULL)
+  name <- info$dbms.name
+  rlang::is_string(name) && identical(tolower(name), "snowflake")
+}
+
+snowflake_table_registry <- function(
+  con,
+  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)
+  )
+}
+
+snowflake_current_namespace <- function(con, call = rlang::caller_env()) {
+  row <- tryCatch(
+    DBI::dbGetQuery(
+      con,
+      paste(
+        "SELECT CURRENT_DATABASE() AS catalog,",
+        "CURRENT_SCHEMA() AS schema"
+      )
+    ),
+    error = function(err) {
+      cli::cli_abort(
+        "Failed to read the current Snowflake namespace.",
+        parent = err,
+        call = call
+      )
+    }
+  )
+  names(row) <- tolower(names(row))
+  if (nrow(row) != 1L || !all(c("catalog", "schema") %in% names(row))) {
+    cli::cli_abort(
+      "Snowflake 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 Snowflake connection has no current database 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"]])
+}
+
+snowflake_list_relations <- function(con, namespace, call = rlang::caller_env()) {
+  components <- namespace@name
+  target <- if (identical(names(components), "catalog")) {
+    paste("IN DATABASE", DBI::dbQuoteIdentifier(con, namespace))
+  } else {
+    paste("IN SCHEMA", DBI::dbQuoteIdentifier(con, namespace))
+  }
+  rows <- tryCatch(
+    DBI::dbGetQuery(con, paste("SHOW OBJECTS", target)),
+    error = function(err) {
+      cli::cli_abort(
+        "Failed to list relations in the selected Snowflake namespace.",
+        parent = err,
+        call = call
+      )
+    }
+  )
+  snowflake_relations_from_show(rows)
+}
+
+snowflake_exact_relation <- function(con, id, call = rlang::caller_env()) {
+  components <- id@name
+  namespace <- components[setdiff(names(components), "table")]
+  if (length(namespace) == 0L) {
+    namespace <- snowflake_current_namespace(con, call = call)@name
+  }
+  namespace <- do.call(DBI::Id, as.list(namespace))
+  target <- paste("IN SCHEMA", DBI::dbQuoteIdentifier(con, namespace))
+  rows <- tryCatch(
+    DBI::dbGetQuery(
+      con,
+      paste(
+        "SHOW OBJECTS LIKE",
+        DBI::dbQuoteString(con, components[["table"]]),
+        target
+      )
+    ),
+    error = function(err) {
+      cli::cli_abort(
+        "Failed to read metadata for the selected Snowflake relation.",
+        parent = err,
+        call = call
+      )
+    }
+  )
+  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
+}
+
+snowflake_relations_from_show <- function(rows) {
+  names(rows) <- tolower(names(rows))
+  if (nrow(rows) == 0L) {
+    return(list())
+  }
+  rows <- rows[toupper(rows$kind) %in% c("TABLE", "VIEW"), , drop = FALSE]
+  lapply(seq_len(nrow(rows)), function(i) {
+    description <- rows$comment[[i]]
+    if (is.na(description) || !nzchar(description)) {
+      description <- NULL
+    }
+    list(
+      id = DBI::Id(
+        catalog = rows$database_name[[i]],
+        schema = rows$schema_name[[i]],
+        table = rows$name[[i]]
+      ),
+      kind = tolower(rows$kind[[i]]),
+      description = description
+    )
+  })
+}
+
+snowflake_describe_relation <- function(con, id, call = rlang::caller_env()) {
+  rows <- tryCatch(
+    DBI::dbGetQuery(
+      con,
+      paste("DESC TABLE", DBI::dbQuoteIdentifier(con, id))
+    ),
+    error = function(err) {
+      cli::cli_abort(
+        "Failed to describe the selected Snowflake relation.",
+        parent = err,
+        call = call
+      )
+    }
+  )
+  names(rows) <- tolower(names(rows))
+  rows <- rows[toupper(rows$kind) == "COLUMN", , drop = FALSE]
+  description <- rows$comment
+  description[is.na(description) | !nzchar(description)] <- NA_character_
+  data.frame(
+    column = rows$name,
+    type = rows$type,
+    nullable = rows[["null?"]] == "Y",
+    description = description,
+    row.names = NULL
+  )
+}
+
+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"
+}

R/data-source.R

@@ -35,7 +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.
+#'   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
+#'   schema.
 #'
 #'   For a board, a named character vector of pins to read: the names become
 #'   table names, and the values are pin names passed to [pins::pin_read()].
@@ -143,6 +146,26 @@ data_source_connection <- function(
 ) {
   span <- local_commons_span("commons_data_source_list_tables")
 
+  if (is_snowflake_connection(con)) {
+    table_registry <- snowflake_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,
+      snowflake_relations = table_registry$relations
+    ))
+  }
+
   if (is.null(tables)) {
     listed <- DBI::dbListTables(con)
     commons_span_set_attribute(span, "commons.data_source.n_tables", length(listed))
@@ -241,7 +264,8 @@ new_data_source <- function(
   owned,
   table_ids = table_ids_from_labels(tables),
   dictionary = NULL,
-  pending = NULL
+  pending = NULL,
+  snowflake_relations = NULL
 ) {
   # Disconnect only the DuckDB connection we created; a user-supplied connection
   # has its own owner and lifetime.
@@ -263,7 +287,8 @@ new_data_source <- function(
       table_ids = table_ids,
       handle = handle,
       dictionary = dictionary,
-      pending = pending
+      pending = pending,
+      snowflake_relations = snowflake_relations
     ),
     class = "commons_data_source"
   )
@@ -426,7 +451,12 @@ pending_tables_in_error <- function(source, err) {
   )]
 }
 
-source_describe <- function(source, table, n_sample = 5) {
+source_describe <- function(
+  source,
+  table,
+  n_sample = 5,
+  call = rlang::caller_env()
+) {
   id <- source$table_ids[[table]]
   if (is.null(id)) {
     cli::cli_abort(c(
@@ -444,12 +474,22 @@ source_describe <- function(source, table, n_sample = 5) {
       n_sample
     )
   )
-  schema <- data.frame(
-    column = names(sample),
-    type = vapply(sample, function(x) class(x)[[1]], character(1)),
-    row.names = NULL
+  relation <- source$snowflake_relations[[table]]
+  if (is.null(source$snowflake_relations)) {
+    schema <- data.frame(
+      column = names(sample),
+      type = vapply(sample, function(x) class(x)[[1]], character(1)),
+      row.names = NULL
+    )
+  } else {
+    schema <- snowflake_describe_relation(source$con, id, call = call)
+  }
+  list(
+    schema = schema,
+    sample = sample,
+    kind = relation$kind,
+    description = relation$description
   )
-  list(schema = schema, sample = sample)
 }
 
 source_query <- function(source, sql) {

R/tools.R

@@ -345,13 +345,18 @@ search_context_tool <- function(context, query) {
 describe_table_tool <- function(source, table, source_name = NULL, tracker = NULL) {
   d <- source_describe(source, table)
   entry <- source$dictionary$tables[[table]]
+  relation <- c(
+    if (!is.null(d$kind)) sprintf("Relation type: %s.", d$kind),
+    d$description
+  )
 
   sample <- sprintf(
     "Sample rows:\n\n%s",
     df_to_markdown(d$sample, max_rows = 5)
   )
   if (is.null(entry)) {
     parts <- c(
+      relation,
       sprintf("Columns of `%s`:\n\n%s", table, df_to_markdown(d$schema)),
       sample
     )
@@ -363,6 +368,7 @@ describe_table_tool <- function(source, table, source_name = NULL, tracker = NUL
       dictionary_columns_text(entry$columns, live = d$schema)
     )
     parts <- c(
+      relation,
       dictionary_entry_parts(source$dictionary, table, columns),
       sample
     )

tests/testthat/README.md

@@ -25,6 +25,8 @@ options(commons.test.databricks = DBI::Id(
 ))
 ```
 
-Each 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 Snowflake test exercises 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.