|
| 1 | +#' @importFrom methods as |
| 2 | +#' |
| 3 | +# Sanitize column name for HDF5 dataset and ensure unique (mirrors Python _safe_column_dataset_name). |
| 4 | +.safe_column_dataset_name <- function(source_name, column_index, existing_names) { |
| 5 | + base <- gsub("[^A-Za-z0-9_.-]", "_", source_name) |
| 6 | + base <- trimws(base) |
| 7 | + base <- sub("^_+|_+$", "", base) |
| 8 | + if (!nzchar(base)) base <- paste0("column_", column_index) |
| 9 | + candidate <- base |
| 10 | + suffix <- 1L |
| 11 | + while (candidate %in% existing_names) { |
| 12 | + candidate <- paste0(base, "_", suffix) |
| 13 | + suffix <- suffix + 1L |
| 14 | + } |
| 15 | + return(candidate) |
| 16 | +} |
| 17 | + |
| 18 | +# Coerce to character, NA -> "" (mirrors Python _as_string_values). |
| 19 | +.as_string_values <- function(x) { |
| 20 | + y <- as.character(x) |
| 21 | + y[is.na(y)] <- "" |
| 22 | + return(y) |
| 23 | +} |
| 24 | + |
| 25 | +# Write optional var (feature) metadata under info/var (mirrors Python _write_var_metadata). |
| 26 | +.write_var_metadata <- function(fid, n_cols, feature_df, feature_names) { |
| 27 | + if (nrow(feature_df) != n_cols) { |
| 28 | + stop("feature_df row count (", nrow(feature_df), ") does not match matrix columns (", n_cols, ").") |
| 29 | + } |
| 30 | + names_source <- if (!is.null(feature_names)) feature_names else rownames(feature_df) |
| 31 | + if (length(names_source) != n_cols) { |
| 32 | + stop("feature_names length (", length(names_source), ") does not match matrix columns (", n_cols, ").") |
| 33 | + } |
| 34 | + rhdf5::h5createGroup(fid, "info") |
| 35 | + rhdf5::h5createGroup(fid, "info/var") |
| 36 | + rhdf5::h5writeDataset(.as_string_values(names_source), h5loc = fid, name = "info/var/var_names", level = 5L) |
| 37 | + rhdf5::h5writeDataset(.as_string_values(rownames(feature_df)), h5loc = fid, name = "info/var/index", level = 5L) |
| 38 | + rhdf5::h5createGroup(fid, "info/var/columns") |
| 39 | + existing <- character(0) |
| 40 | + for (i in seq_len(ncol(feature_df))) { |
| 41 | + col_name <- names(feature_df)[i] |
| 42 | + dataset_name <- .safe_column_dataset_name(col_name, i - 1L, existing) |
| 43 | + existing <- c(existing, dataset_name) |
| 44 | + col_path <- paste0("info/var/columns/", dataset_name) |
| 45 | + vec <- feature_df[[i]] |
| 46 | + if (is.factor(vec)) vec <- as.character(vec) |
| 47 | + if (is.logical(vec)) { |
| 48 | + storage.mode(vec) <- "integer" |
| 49 | + if (any(is.na(vec))) vec[is.na(vec)] <- -1L |
| 50 | + rhdf5::h5writeDataset(vec, h5loc = fid, name = col_path, level = 5L) |
| 51 | + } else if (is.numeric(vec)) { |
| 52 | + rhdf5::h5writeDataset(as.double(vec), h5loc = fid, name = col_path, level = 5L) |
| 53 | + } else { |
| 54 | + rhdf5::h5writeDataset(.as_string_values(vec), h5loc = fid, name = col_path, level = 5L) |
| 55 | + } |
| 56 | + } |
| 57 | + invisible(NULL) |
| 58 | +} |
| 59 | + |
| 60 | +.save_vars_h5 <- function(out_file, mat, feature_df = NULL, feature_names = NULL, |
| 61 | + col_batch = NULL, min_chunk_size = 10000L) { |
| 62 | + m <- as(mat, "CsparseMatrix") |
| 63 | + n_obs <- nrow(m) |
| 64 | + n_vars <- ncol(m) |
| 65 | + |
| 66 | + if (is.null(col_batch)) { |
| 67 | + col_batch <- max(1L, as.integer(100000000 / max(n_obs, 1))) |
| 68 | + } |
| 69 | + chunk_size <- max(1L, min(n_obs * 10L, min_chunk_size)) |
| 70 | + |
| 71 | + if (file.exists(out_file) && !file.remove(out_file)) { |
| 72 | + stop("Could not remove existing file: ", out_file) |
| 73 | + } |
| 74 | + |
| 75 | + rhdf5::h5createFile(out_file) |
| 76 | + fid <- rhdf5::H5Fopen(out_file, flags = "H5F_ACC_RDWR") |
| 77 | + on.exit(rhdf5::H5Fclose(fid), add = TRUE) |
| 78 | + |
| 79 | + rhdf5::h5createGroup(fid, "vars") |
| 80 | + rhdf5::h5writeAttribute(as.integer(n_obs), h5obj = fid, name = "vars/n_obs") |
| 81 | + rhdf5::h5writeAttribute(as.integer(n_vars), h5obj = fid, name = "vars/n_vars") |
| 82 | + |
| 83 | + # Create extensible datasets (equivalent to maxshape=(None,) in h5py) |
| 84 | + max_nnz <- n_obs * n_vars # upper bound |
| 85 | + rhdf5::h5createDataset( |
| 86 | + fid, "vars/indices", |
| 87 | + dims = 0L, |
| 88 | + maxdims = rhdf5::H5Sunlimited(), |
| 89 | + chunk = chunk_size, |
| 90 | + H5type = "H5T_NATIVE_INT32", |
| 91 | + level = 0L # LZ4 not directly available, use level for deflate or 0 for none |
| 92 | + ) |
| 93 | + rhdf5::h5createDataset( |
| 94 | + fid, "vars/data", |
| 95 | + dims = 0L, |
| 96 | + maxdims = rhdf5::H5Sunlimited(), |
| 97 | + chunk = chunk_size, |
| 98 | + H5type = "H5T_NATIVE_FLOAT", |
| 99 | + level = 5L |
| 100 | + ) |
| 101 | + |
| 102 | + indptr <- 0L |
| 103 | + current_size <- 0L |
| 104 | + |
| 105 | + starts <- seq(1L, n_vars, by = col_batch) |
| 106 | + for (start in starts) { |
| 107 | + end <- min(start + col_batch - 1L, n_vars) |
| 108 | + chunk <- as(m[, start:end, drop = FALSE], "CsparseMatrix") |
| 109 | + |
| 110 | + chunk_indices <- as.integer(chunk@i) |
| 111 | + chunk_data <- as.numeric(chunk@x) |
| 112 | + chunk_nnz <- length(chunk_indices) |
| 113 | + |
| 114 | + if (chunk_nnz > 0L) { |
| 115 | + # Extend and write indices |
| 116 | + rhdf5::h5set_extent(fid, "vars/indices", current_size + chunk_nnz) |
| 117 | + rhdf5::h5writeDataset( |
| 118 | + chunk_indices, h5loc = fid, name = "vars/indices", |
| 119 | + index = list((current_size + 1L):(current_size + chunk_nnz)) |
| 120 | + ) |
| 121 | + # Extend and write data |
| 122 | + rhdf5::h5set_extent(fid, "vars/data", current_size + chunk_nnz) |
| 123 | + rhdf5::h5writeDataset( |
| 124 | + chunk_data, h5loc = fid, name = "vars/data", |
| 125 | + index = list((current_size + 1L):(current_size + chunk_nnz)) |
| 126 | + ) |
| 127 | + current_size <- current_size + chunk_nnz |
| 128 | + } |
| 129 | + |
| 130 | + # Accumulate indptr (skip first element after first chunk) |
| 131 | + new_indptr <- as.integer(chunk@p[-1L] + indptr[length(indptr)]) |
| 132 | + indptr <- c(indptr, new_indptr) |
| 133 | + } |
| 134 | + |
| 135 | + rhdf5::h5writeDataset(as.integer(indptr), h5loc = fid, name = "vars/indptr", level = 5L) |
| 136 | + |
| 137 | + if (!is.null(feature_df)) { |
| 138 | + .write_var_metadata(fid, n_cols = n_vars, feature_df = feature_df, feature_names = feature_names) |
| 139 | + } |
| 140 | + |
| 141 | + invisible(out_file) |
| 142 | +} |
| 143 | + |
| 144 | +.save_obs_duckdb <- function(out_file, obs_df, table_name = "obs", |
| 145 | + threads = "4", memory_limit = "4GB", temp_directory = "tmp/duckdb") { |
| 146 | + if (!requireNamespace("duckdb", quietly = TRUE)) { |
| 147 | + stop("Package 'duckdb' is required to build obs.duckdb. Install with: install.packages('duckdb')") |
| 148 | + } |
| 149 | + if (!grepl("^[A-Za-z_][A-Za-z0-9_]*$", table_name)) { |
| 150 | + stop("Invalid table_name. Use letters, numbers, and underscores only.") |
| 151 | + } |
| 152 | + if (file.exists(out_file)) file.remove(out_file) |
| 153 | + config <- list(threads = as.character(threads), memory_limit = memory_limit, temp_directory = temp_directory) |
| 154 | + con <- duckdb::dbConnect(duckdb::duckdb(), out_file, config = config) |
| 155 | + on.exit(duckdb::dbDisconnect(con, shutdown = TRUE), add = TRUE) |
| 156 | + duckdb::dbWriteTable(con, table_name, as.data.frame(obs_df), overwrite = TRUE) |
| 157 | + invisible(out_file) |
| 158 | +} |
0 commit comments