| 1 | /* |
| 2 | * Copyright 2020 Google LLC |
| 3 | * |
| 4 | * Use of this source code is governed by a BSD-style |
| 5 | * license that can be found in the LICENSE file or at |
| 6 | * https://developers.google.com/open-source/licenses/bsd |
| 7 | */ |
| 8 | |
| 9 | #include "merged.h" |
| 10 | |
| 11 | #include "constants.h" |
| 12 | #include "iter.h" |
| 13 | #include "pq.h" |
| 14 | #include "record.h" |
| 15 | #include "reftable-merged.h" |
| 16 | #include "reftable-error.h" |
| 17 | #include "system.h" |
| 18 | #include "table.h" |
| 19 | |
| 20 | struct merged_subiter { |
| 21 | struct reftable_iterator iter; |
| 22 | struct reftable_record rec; |
| 23 | }; |
| 24 | |
| 25 | struct merged_iter { |
| 26 | struct merged_subiter *subiters; |
| 27 | struct merged_iter_pqueue pq; |
| 28 | size_t subiters_len; |
| 29 | int suppress_deletions; |
| 30 | ssize_t advance_index; |
| 31 | }; |
| 32 | |
| 33 | static void merged_iter_close(void *p) |
| 34 | { |
| 35 | struct merged_iter *mi = p; |
| 36 | |
| 37 | merged_iter_pqueue_release(&mi->pq); |
| 38 | for (size_t i = 0; i < mi->subiters_len; i++) { |
| 39 | reftable_iterator_destroy(&mi->subiters[i].iter); |
| 40 | reftable_record_release(&mi->subiters[i].rec); |
| 41 | } |
| 42 | reftable_free(mi->subiters); |
| 43 | } |
| 44 | |
| 45 | static int merged_iter_advance_subiter(struct merged_iter *mi, size_t idx) |
| 46 | { |
| 47 | struct pq_entry e = { |
| 48 | .index = idx, |
| 49 | .rec = &mi->subiters[idx].rec, |
| 50 | }; |
| 51 | int err; |
| 52 | |
| 53 | err = iterator_next(&mi->subiters[idx].iter, &mi->subiters[idx].rec); |
| 54 | if (err) |
| 55 | return err; |
| 56 | |
| 57 | err = merged_iter_pqueue_add(&mi->pq, &e); |
| 58 | if (err) |
| 59 | return err; |
| 60 | |
| 61 | return 0; |
| 62 | } |
| 63 | |
| 64 | static int merged_iter_seek(struct merged_iter *mi, struct reftable_record *want) |
| 65 | { |
| 66 | int err; |
| 67 | |
| 68 | mi->advance_index = -1; |
| 69 | while (!merged_iter_pqueue_is_empty(mi->pq)) { |
| 70 | err = merged_iter_pqueue_remove(&mi->pq, NULL); |
| 71 | if (err < 0) |
| 72 | return err; |
| 73 | } |
| 74 | |
| 75 | for (size_t i = 0; i < mi->subiters_len; i++) { |
| 76 | err = iterator_seek(&mi->subiters[i].iter, want); |
| 77 | if (err < 0) |
| 78 | return err; |
| 79 | if (err > 0) |
| 80 | continue; |
| 81 | |
| 82 | err = merged_iter_advance_subiter(mi, i); |
| 83 | if (err < 0) |
| 84 | return err; |
| 85 | } |
| 86 | |
| 87 | return 0; |
| 88 | } |
| 89 | |
| 90 | static int merged_iter_next_entry(struct merged_iter *mi, |
| 91 | struct reftable_record *rec) |
| 92 | { |
| 93 | struct pq_entry entry = { 0 }; |
| 94 | int err = 0, empty; |
| 95 | |
| 96 | empty = merged_iter_pqueue_is_empty(mi->pq); |
| 97 | |
| 98 | if (mi->advance_index >= 0) { |
| 99 | /* |
| 100 | * When there are no pqueue entries then we only have a single |
| 101 | * subiter left. There is no need to use the pqueue in that |
| 102 | * case anymore as we know that the subiter will return entries |
| 103 | * in the correct order already. |
| 104 | * |
| 105 | * While this may sound like a very specific edge case, it may |
| 106 | * happen more frequently than you think. Most repositories |
| 107 | * will end up having a single large base table that contains |
| 108 | * most of the refs. It's thus likely that we exhaust all |
| 109 | * subiters but the one from that base ref. |
| 110 | */ |
| 111 | if (empty) |
| 112 | return iterator_next(&mi->subiters[mi->advance_index].iter, |
| 113 | rec); |
| 114 | |
| 115 | err = merged_iter_advance_subiter(mi, mi->advance_index); |
| 116 | if (err < 0) |
| 117 | return err; |
| 118 | if (!err) |
| 119 | empty = 0; |
| 120 | mi->advance_index = -1; |
| 121 | } |
| 122 | |
| 123 | if (empty) |
| 124 | return 1; |
| 125 | |
| 126 | err = merged_iter_pqueue_remove(&mi->pq, &entry); |
| 127 | if (err < 0) |
| 128 | return err; |
| 129 | |
| 130 | /* |
| 131 | One can also use reftable as datacenter-local storage, where the ref |
| 132 | database is maintained in globally consistent database (eg. |
| 133 | CockroachDB or Spanner). In this scenario, replication delays together |
| 134 | with compaction may cause newer tables to contain older entries. In |
| 135 | such a deployment, the loop below must be changed to collect all |
| 136 | entries for the same key, and return new the newest one. |
| 137 | */ |
| 138 | while (!merged_iter_pqueue_is_empty(mi->pq)) { |
| 139 | struct pq_entry top = merged_iter_pqueue_top(mi->pq); |
| 140 | int cmp; |
| 141 | |
| 142 | err = reftable_record_cmp(top.rec, entry.rec, &cmp); |
| 143 | if (err < 0) |
| 144 | return err; |
| 145 | if (cmp > 0) |
| 146 | break; |
| 147 | |
| 148 | err = merged_iter_pqueue_remove(&mi->pq, NULL); |
| 149 | if (err < 0) |
| 150 | return err; |
| 151 | |
| 152 | err = merged_iter_advance_subiter(mi, top.index); |
| 153 | if (err < 0) |
| 154 | return err; |
| 155 | } |
| 156 | |
| 157 | mi->advance_index = entry.index; |
| 158 | REFTABLE_SWAP(*rec, *entry.rec); |
| 159 | return 0; |
| 160 | } |
| 161 | |
| 162 | static int merged_iter_seek_void(void *it, struct reftable_record *want) |
| 163 | { |
| 164 | return merged_iter_seek(it, want); |
| 165 | } |
| 166 | |
| 167 | static int merged_iter_next_void(void *p, struct reftable_record *rec) |
| 168 | { |
| 169 | struct merged_iter *mi = p; |
| 170 | while (1) { |
| 171 | int err = merged_iter_next_entry(mi, rec); |
| 172 | if (err) |
| 173 | return err; |
| 174 | if (mi->suppress_deletions && reftable_record_is_deletion(rec)) |
| 175 | continue; |
| 176 | return 0; |
| 177 | } |
| 178 | } |
| 179 | |
| 180 | static struct reftable_iterator_vtable merged_iter_vtable = { |
| 181 | .seek = merged_iter_seek_void, |
| 182 | .next = &merged_iter_next_void, |
| 183 | .close = &merged_iter_close, |
| 184 | }; |
| 185 | |
| 186 | static void iterator_from_merged_iter(struct reftable_iterator *it, |
| 187 | struct merged_iter *mi) |
| 188 | { |
| 189 | assert(!it->ops); |
| 190 | it->iter_arg = mi; |
| 191 | it->ops = &merged_iter_vtable; |
| 192 | } |
| 193 | |
| 194 | int reftable_merged_table_new(struct reftable_merged_table **dest, |
| 195 | struct reftable_table **tables, size_t n, |
| 196 | enum reftable_hash hash_id) |
| 197 | { |
| 198 | struct reftable_merged_table *m = NULL; |
| 199 | uint64_t last_max = 0; |
| 200 | uint64_t first_min = 0; |
| 201 | |
| 202 | for (size_t i = 0; i < n; i++) { |
| 203 | uint64_t min = reftable_table_min_update_index(tables[i]); |
| 204 | uint64_t max = reftable_table_max_update_index(tables[i]); |
| 205 | |
| 206 | if (reftable_table_hash_id(tables[i]) != hash_id) { |
| 207 | return REFTABLE_FORMAT_ERROR; |
| 208 | } |
| 209 | if (i == 0 || min < first_min) { |
| 210 | first_min = min; |
| 211 | } |
| 212 | if (i == 0 || max > last_max) { |
| 213 | last_max = max; |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | REFTABLE_CALLOC_ARRAY(m, 1); |
| 218 | if (!m) |
| 219 | return REFTABLE_OUT_OF_MEMORY_ERROR; |
| 220 | |
| 221 | m->tables = tables; |
| 222 | m->tables_len = n; |
| 223 | m->min = first_min; |
| 224 | m->max = last_max; |
| 225 | m->hash_id = hash_id; |
| 226 | *dest = m; |
| 227 | return 0; |
| 228 | } |
| 229 | |
| 230 | void reftable_merged_table_free(struct reftable_merged_table *mt) |
| 231 | { |
| 232 | if (!mt) |
| 233 | return; |
| 234 | reftable_free(mt); |
| 235 | } |
| 236 | |
| 237 | uint64_t |
| 238 | reftable_merged_table_max_update_index(struct reftable_merged_table *mt) |
| 239 | { |
| 240 | return mt->max; |
| 241 | } |
| 242 | |
| 243 | uint64_t |
| 244 | reftable_merged_table_min_update_index(struct reftable_merged_table *mt) |
| 245 | { |
| 246 | return mt->min; |
| 247 | } |
| 248 | |
| 249 | int merged_table_init_iter(struct reftable_merged_table *mt, |
| 250 | struct reftable_iterator *it, |
| 251 | uint8_t typ) |
| 252 | { |
| 253 | struct merged_subiter *subiters = NULL; |
| 254 | struct merged_iter *mi = NULL; |
| 255 | int ret; |
| 256 | |
| 257 | if (mt->tables_len) { |
| 258 | REFTABLE_CALLOC_ARRAY(subiters, mt->tables_len); |
| 259 | if (!subiters) { |
| 260 | ret = REFTABLE_OUT_OF_MEMORY_ERROR; |
| 261 | goto out; |
| 262 | } |
| 263 | } |
| 264 | |
| 265 | for (size_t i = 0; i < mt->tables_len; i++) { |
| 266 | ret = reftable_record_init(&subiters[i].rec, typ); |
| 267 | if (ret < 0) |
| 268 | goto out; |
| 269 | |
| 270 | ret = table_init_iter(mt->tables[i], &subiters[i].iter, typ); |
| 271 | if (ret < 0) |
| 272 | goto out; |
| 273 | } |
| 274 | |
| 275 | REFTABLE_CALLOC_ARRAY(mi, 1); |
| 276 | if (!mi) { |
| 277 | ret = REFTABLE_OUT_OF_MEMORY_ERROR; |
| 278 | goto out; |
| 279 | } |
| 280 | mi->advance_index = -1; |
| 281 | mi->suppress_deletions = mt->suppress_deletions; |
| 282 | mi->subiters = subiters; |
| 283 | mi->subiters_len = mt->tables_len; |
| 284 | |
| 285 | iterator_from_merged_iter(it, mi); |
| 286 | ret = 0; |
| 287 | |
| 288 | out: |
| 289 | if (ret < 0) { |
| 290 | for (size_t i = 0; subiters && i < mt->tables_len; i++) { |
| 291 | reftable_iterator_destroy(&subiters[i].iter); |
| 292 | reftable_record_release(&subiters[i].rec); |
| 293 | } |
| 294 | reftable_free(subiters); |
| 295 | reftable_free(mi); |
| 296 | } |
| 297 | |
| 298 | return ret; |
| 299 | } |
| 300 | |
| 301 | int reftable_merged_table_init_ref_iterator(struct reftable_merged_table *mt, |
| 302 | struct reftable_iterator *it) |
| 303 | { |
| 304 | return merged_table_init_iter(mt, it, REFTABLE_BLOCK_TYPE_REF); |
| 305 | } |
| 306 | |
| 307 | int reftable_merged_table_init_log_iterator(struct reftable_merged_table *mt, |
| 308 | struct reftable_iterator *it) |
| 309 | { |
| 310 | return merged_table_init_iter(mt, it, REFTABLE_BLOCK_TYPE_LOG); |
| 311 | } |
| 312 | |
| 313 | enum reftable_hash reftable_merged_table_hash_id(struct reftable_merged_table *mt) |
| 314 | { |
| 315 | return mt->hash_id; |
| 316 | } |