reftable/merged.c

   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 "reader.h"
  15 #include "record.h"
  16 #include "reftable-merged.h"
  17 #include "reftable-error.h"
  18 #include "system.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
  70         for (size_t i = 0; i < mi->subiters_len; i++) {
  71                 err = iterator_seek(&mi->subiters[i].iter, want);
  72                 if (err < 0)
  73                         return err;
  74                 if (err > 0)
  75                         continue;
  76
  77                 err = merged_iter_advance_subiter(mi, i);
  78                 if (err < 0)
  79                         return err;
  80         }
  81
  82         return 0;
  83 }
  84
  85 static int merged_iter_next_entry(struct merged_iter *mi,
  86                                   struct reftable_record *rec)
  87 {
  88         struct pq_entry entry = { 0 };
  89         int err = 0, empty;
  90
  91         empty = merged_iter_pqueue_is_empty(mi->pq);
  92
  93         if (mi->advance_index >= 0) {
  94                 /*
  95                  * When there are no pqueue entries then we only have a single
  96                  * subiter left. There is no need to use the pqueue in that
  97                  * case anymore as we know that the subiter will return entries
  98                  * in the correct order already.
  99                  *
 100                  * While this may sound like a very specific edge case, it may
 101                  * happen more frequently than you think. Most repositories
 102                  * will end up having a single large base table that contains
 103                  * most of the refs. It's thus likely that we exhaust all
 104                  * subiters but the one from that base ref.
 105                  */
 106                 if (empty)
 107                         return iterator_next(&mi->subiters[mi->advance_index].iter,
 108                                              rec);
 109
 110                 err = merged_iter_advance_subiter(mi, mi->advance_index);
 111                 if (err < 0)
 112                         return err;
 113                 if (!err)
 114                         empty = 0;
 115                 mi->advance_index = -1;
 116         }
 117
 118         if (empty)
 119                 return 1;
 120
 121         entry = merged_iter_pqueue_remove(&mi->pq);
 122
 123         /*
 124           One can also use reftable as datacenter-local storage, where the ref
 125           database is maintained in globally consistent database (eg.
 126           CockroachDB or Spanner). In this scenario, replication delays together
 127           with compaction may cause newer tables to contain older entries. In
 128           such a deployment, the loop below must be changed to collect all
 129           entries for the same key, and return new the newest one.
 130         */
 131         while (!merged_iter_pqueue_is_empty(mi->pq)) {
 132                 struct pq_entry top = merged_iter_pqueue_top(mi->pq);
 133                 int cmp;
 134
 135                 cmp = reftable_record_cmp(top.rec, entry.rec);
 136                 if (cmp > 0)
 137                         break;
 138
 139                 merged_iter_pqueue_remove(&mi->pq);
 140                 err = merged_iter_advance_subiter(mi, top.index);
 141                 if (err < 0)
 142                         return err;
 143         }
 144
 145         mi->advance_index = entry.index;
 146         SWAP(*rec, *entry.rec);
 147         return 0;
 148 }
 149
 150 static int merged_iter_seek_void(void *it, struct reftable_record *want)
 151 {
 152         return merged_iter_seek(it, want);
 153 }
 154
 155 static int merged_iter_next_void(void *p, struct reftable_record *rec)
 156 {
 157         struct merged_iter *mi = p;
 158         while (1) {
 159                 int err = merged_iter_next_entry(mi, rec);
 160                 if (err)
 161                         return err;
 162                 if (mi->suppress_deletions && reftable_record_is_deletion(rec))
 163                         continue;
 164                 return 0;
 165         }
 166 }
 167
 168 static struct reftable_iterator_vtable merged_iter_vtable = {
 169         .seek = merged_iter_seek_void,
 170         .next = &merged_iter_next_void,
 171         .close = &merged_iter_close,
 172 };
 173
 174 static void iterator_from_merged_iter(struct reftable_iterator *it,
 175                                       struct merged_iter *mi)
 176 {
 177         assert(!it->ops);
 178         it->iter_arg = mi;
 179         it->ops = &merged_iter_vtable;
 180 }
 181
 182 int reftable_merged_table_new(struct reftable_merged_table **dest,
 183                               struct reftable_reader **readers, size_t n,
 184                               uint32_t hash_id)
 185 {
 186         struct reftable_merged_table *m = NULL;
 187         uint64_t last_max = 0;
 188         uint64_t first_min = 0;
 189
 190         for (size_t i = 0; i < n; i++) {
 191                 uint64_t min = reftable_reader_min_update_index(readers[i]);
 192                 uint64_t max = reftable_reader_max_update_index(readers[i]);
 193
 194                 if (reftable_reader_hash_id(readers[i]) != hash_id) {
 195                         return REFTABLE_FORMAT_ERROR;
 196                 }
 197                 if (i == 0 || min < first_min) {
 198                         first_min = min;
 199                 }
 200                 if (i == 0 || max > last_max) {
 201                         last_max = max;
 202                 }
 203         }
 204
 205         REFTABLE_CALLOC_ARRAY(m, 1);
 206         if (!m)
 207                 return REFTABLE_OUT_OF_MEMORY_ERROR;
 208
 209         m->readers = readers;
 210         m->readers_len = n;
 211         m->min = first_min;
 212         m->max = last_max;
 213         m->hash_id = hash_id;
 214         *dest = m;
 215         return 0;
 216 }
 217
 218 void reftable_merged_table_free(struct reftable_merged_table *mt)
 219 {
 220         if (!mt)
 221                 return;
 222         reftable_free(mt);
 223 }
 224
 225 uint64_t
 226 reftable_merged_table_max_update_index(struct reftable_merged_table *mt)
 227 {
 228         return mt->max;
 229 }
 230
 231 uint64_t
 232 reftable_merged_table_min_update_index(struct reftable_merged_table *mt)
 233 {
 234         return mt->min;
 235 }
 236
 237 int merged_table_init_iter(struct reftable_merged_table *mt,
 238                            struct reftable_iterator *it,
 239                            uint8_t typ)
 240 {
 241         struct merged_subiter *subiters;
 242         struct merged_iter *mi = NULL;
 243         int ret;
 244
 245         REFTABLE_CALLOC_ARRAY(subiters, mt->readers_len);
 246         if (!subiters) {
 247                 ret = REFTABLE_OUT_OF_MEMORY_ERROR;
 248                 goto out;
 249         }
 250
 251         for (size_t i = 0; i < mt->readers_len; i++) {
 252                 reftable_record_init(&subiters[i].rec, typ);
 253                 ret = reader_init_iter(mt->readers[i], &subiters[i].iter, typ);
 254                 if (ret < 0)
 255                         goto out;
 256         }
 257
 258         REFTABLE_CALLOC_ARRAY(mi, 1);
 259         if (!mi) {
 260                 ret = REFTABLE_OUT_OF_MEMORY_ERROR;
 261                 goto out;
 262         }
 263         mi->advance_index = -1;
 264         mi->suppress_deletions = mt->suppress_deletions;
 265         mi->subiters = subiters;
 266         mi->subiters_len = mt->readers_len;
 267
 268         iterator_from_merged_iter(it, mi);
 269         ret = 0;
 270
 271 out:
 272         if (ret < 0) {
 273                 for (size_t i = 0; subiters && i < mt->readers_len; i++) {
 274                         reftable_iterator_destroy(&subiters[i].iter);
 275                         reftable_record_release(&subiters[i].rec);
 276                 }
 277                 reftable_free(subiters);
 278                 reftable_free(mi);
 279         }
 280
 281         return ret;
 282 }
 283
 284 int reftable_merged_table_init_ref_iterator(struct reftable_merged_table *mt,
 285                                             struct reftable_iterator *it)
 286 {
 287         return merged_table_init_iter(mt, it, BLOCK_TYPE_REF);
 288 }
 289
 290 int reftable_merged_table_init_log_iterator(struct reftable_merged_table *mt,
 291                                             struct reftable_iterator *it)
 292 {
 293         return merged_table_init_iter(mt, it, BLOCK_TYPE_LOG);
 294 }
 295
 296 uint32_t reftable_merged_table_hash_id(struct reftable_merged_table *mt)
 297 {
 298         return mt->hash_id;
 299 }