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 "record.h"
  15 #include "generic.h"
  16 #include "reftable-merged.h"
  17 #include "reftable-error.h"
  18 #include "system.h"
  19
  20 static int merged_iter_init(struct merged_iter *mi)
  21 {
  22         int i = 0;
  23         for (i = 0; i < mi->stack_len; i++) {
  24                 struct reftable_record rec = reftable_new_record(mi->typ);
  25                 int err = iterator_next(&mi->stack[i], &rec);
  26                 if (err < 0) {
  27                         return err;
  28                 }
  29
  30                 if (err > 0) {
  31                         reftable_iterator_destroy(&mi->stack[i]);
  32                         reftable_record_release(&rec);
  33                 } else {
  34                         struct pq_entry e = {
  35                                 .rec = rec,
  36                                 .index = i,
  37                         };
  38                         merged_iter_pqueue_add(&mi->pq, &e);
  39                 }
  40         }
  41
  42         return 0;
  43 }
  44
  45 static void merged_iter_close(void *p)
  46 {
  47         struct merged_iter *mi = p;
  48         int i = 0;
  49         merged_iter_pqueue_release(&mi->pq);
  50         for (i = 0; i < mi->stack_len; i++) {
  51                 reftable_iterator_destroy(&mi->stack[i]);
  52         }
  53         reftable_free(mi->stack);
  54         strbuf_release(&mi->key);
  55         strbuf_release(&mi->entry_key);
  56 }
  57
  58 static int merged_iter_advance_nonnull_subiter(struct merged_iter *mi,
  59                                                size_t idx)
  60 {
  61         struct pq_entry e = {
  62                 .rec = reftable_new_record(mi->typ),
  63                 .index = idx,
  64         };
  65         int err = iterator_next(&mi->stack[idx], &e.rec);
  66         if (err < 0)
  67                 return err;
  68
  69         if (err > 0) {
  70                 reftable_iterator_destroy(&mi->stack[idx]);
  71                 reftable_record_release(&e.rec);
  72                 return 0;
  73         }
  74
  75         merged_iter_pqueue_add(&mi->pq, &e);
  76         return 0;
  77 }
  78
  79 static int merged_iter_advance_subiter(struct merged_iter *mi, size_t idx)
  80 {
  81         if (iterator_is_null(&mi->stack[idx]))
  82                 return 0;
  83         return merged_iter_advance_nonnull_subiter(mi, idx);
  84 }
  85
  86 static int merged_iter_next_entry(struct merged_iter *mi,
  87                                   struct reftable_record *rec)
  88 {
  89         struct pq_entry entry = { 0 };
  90         int err = 0;
  91
  92         if (merged_iter_pqueue_is_empty(mi->pq))
  93                 return 1;
  94
  95         entry = merged_iter_pqueue_remove(&mi->pq);
  96         err = merged_iter_advance_subiter(mi, entry.index);
  97         if (err < 0)
  98                 return err;
  99
 100         /*
 101           One can also use reftable as datacenter-local storage, where the ref
 102           database is maintained in globally consistent database (eg.
 103           CockroachDB or Spanner). In this scenario, replication delays together
 104           with compaction may cause newer tables to contain older entries. In
 105           such a deployment, the loop below must be changed to collect all
 106           entries for the same key, and return new the newest one.
 107         */
 108         reftable_record_key(&entry.rec, &mi->entry_key);
 109         while (!merged_iter_pqueue_is_empty(mi->pq)) {
 110                 struct pq_entry top = merged_iter_pqueue_top(mi->pq);
 111                 int cmp = 0;
 112
 113                 reftable_record_key(&top.rec, &mi->key);
 114
 115                 cmp = strbuf_cmp(&mi->key, &mi->entry_key);
 116                 if (cmp > 0)
 117                         break;
 118
 119                 merged_iter_pqueue_remove(&mi->pq);
 120                 err = merged_iter_advance_subiter(mi, top.index);
 121                 if (err < 0)
 122                         goto done;
 123                 reftable_record_release(&top.rec);
 124         }
 125
 126         reftable_record_release(rec);
 127         *rec = entry.rec;
 128
 129 done:
 130         if (err)
 131                 reftable_record_release(&entry.rec);
 132         return err;
 133 }
 134
 135 static int merged_iter_next(struct merged_iter *mi, struct reftable_record *rec)
 136 {
 137         while (1) {
 138                 int err = merged_iter_next_entry(mi, rec);
 139                 if (err == 0 && mi->suppress_deletions &&
 140                     reftable_record_is_deletion(rec)) {
 141                         continue;
 142                 }
 143
 144                 return err;
 145         }
 146 }
 147
 148 static int merged_iter_next_void(void *p, struct reftable_record *rec)
 149 {
 150         struct merged_iter *mi = p;
 151         if (merged_iter_pqueue_is_empty(mi->pq))
 152                 return 1;
 153
 154         return merged_iter_next(mi, rec);
 155 }
 156
 157 static struct reftable_iterator_vtable merged_iter_vtable = {
 158         .next = &merged_iter_next_void,
 159         .close = &merged_iter_close,
 160 };
 161
 162 static void iterator_from_merged_iter(struct reftable_iterator *it,
 163                                       struct merged_iter *mi)
 164 {
 165         assert(!it->ops);
 166         it->iter_arg = mi;
 167         it->ops = &merged_iter_vtable;
 168 }
 169
 170 int reftable_new_merged_table(struct reftable_merged_table **dest,
 171                               struct reftable_table *stack, int n,
 172                               uint32_t hash_id)
 173 {
 174         struct reftable_merged_table *m = NULL;
 175         uint64_t last_max = 0;
 176         uint64_t first_min = 0;
 177         int i = 0;
 178         for (i = 0; i < n; i++) {
 179                 uint64_t min = reftable_table_min_update_index(&stack[i]);
 180                 uint64_t max = reftable_table_max_update_index(&stack[i]);
 181
 182                 if (reftable_table_hash_id(&stack[i]) != hash_id) {
 183                         return REFTABLE_FORMAT_ERROR;
 184                 }
 185                 if (i == 0 || min < first_min) {
 186                         first_min = min;
 187                 }
 188                 if (i == 0 || max > last_max) {
 189                         last_max = max;
 190                 }
 191         }
 192
 193         m = reftable_calloc(sizeof(struct reftable_merged_table));
 194         m->stack = stack;
 195         m->stack_len = n;
 196         m->min = first_min;
 197         m->max = last_max;
 198         m->hash_id = hash_id;
 199         *dest = m;
 200         return 0;
 201 }
 202
 203 /* clears the list of subtable, without affecting the readers themselves. */
 204 void merged_table_release(struct reftable_merged_table *mt)
 205 {
 206         FREE_AND_NULL(mt->stack);
 207         mt->stack_len = 0;
 208 }
 209
 210 void reftable_merged_table_free(struct reftable_merged_table *mt)
 211 {
 212         if (!mt) {
 213                 return;
 214         }
 215         merged_table_release(mt);
 216         reftable_free(mt);
 217 }
 218
 219 uint64_t
 220 reftable_merged_table_max_update_index(struct reftable_merged_table *mt)
 221 {
 222         return mt->max;
 223 }
 224
 225 uint64_t
 226 reftable_merged_table_min_update_index(struct reftable_merged_table *mt)
 227 {
 228         return mt->min;
 229 }
 230
 231 static int reftable_table_seek_record(struct reftable_table *tab,
 232                                       struct reftable_iterator *it,
 233                                       struct reftable_record *rec)
 234 {
 235         return tab->ops->seek_record(tab->table_arg, it, rec);
 236 }
 237
 238 static int merged_table_seek_record(struct reftable_merged_table *mt,
 239                                     struct reftable_iterator *it,
 240                                     struct reftable_record *rec)
 241 {
 242         struct reftable_iterator *iters = reftable_calloc(
 243                 sizeof(struct reftable_iterator) * mt->stack_len);
 244         struct merged_iter merged = {
 245                 .stack = iters,
 246                 .typ = reftable_record_type(rec),
 247                 .hash_id = mt->hash_id,
 248                 .suppress_deletions = mt->suppress_deletions,
 249                 .key = STRBUF_INIT,
 250                 .entry_key = STRBUF_INIT,
 251         };
 252         int n = 0;
 253         int err = 0;
 254         int i = 0;
 255         for (i = 0; i < mt->stack_len && err == 0; i++) {
 256                 int e = reftable_table_seek_record(&mt->stack[i], &iters[n],
 257                                                    rec);
 258                 if (e < 0) {
 259                         err = e;
 260                 }
 261                 if (e == 0) {
 262                         n++;
 263                 }
 264         }
 265         if (err < 0) {
 266                 int i = 0;
 267                 for (i = 0; i < n; i++) {
 268                         reftable_iterator_destroy(&iters[i]);
 269                 }
 270                 reftable_free(iters);
 271                 return err;
 272         }
 273
 274         merged.stack_len = n;
 275         err = merged_iter_init(&merged);
 276         if (err < 0) {
 277                 merged_iter_close(&merged);
 278                 return err;
 279         } else {
 280                 struct merged_iter *p =
 281                         reftable_malloc(sizeof(struct merged_iter));
 282                 *p = merged;
 283                 iterator_from_merged_iter(it, p);
 284         }
 285         return 0;
 286 }
 287
 288 int reftable_merged_table_seek_ref(struct reftable_merged_table *mt,
 289                                    struct reftable_iterator *it,
 290                                    const char *name)
 291 {
 292         struct reftable_record rec = {
 293                 .type = BLOCK_TYPE_REF,
 294                 .u.ref = {
 295                         .refname = (char *)name,
 296                 },
 297         };
 298         return merged_table_seek_record(mt, it, &rec);
 299 }
 300
 301 int reftable_merged_table_seek_log_at(struct reftable_merged_table *mt,
 302                                       struct reftable_iterator *it,
 303                                       const char *name, uint64_t update_index)
 304 {
 305         struct reftable_record rec = { .type = BLOCK_TYPE_LOG,
 306                                        .u.log = {
 307                                                .refname = (char *)name,
 308                                                .update_index = update_index,
 309                                        } };
 310         return merged_table_seek_record(mt, it, &rec);
 311 }
 312
 313 int reftable_merged_table_seek_log(struct reftable_merged_table *mt,
 314                                    struct reftable_iterator *it,
 315                                    const char *name)
 316 {
 317         uint64_t max = ~((uint64_t)0);
 318         return reftable_merged_table_seek_log_at(mt, it, name, max);
 319 }
 320
 321 uint32_t reftable_merged_table_hash_id(struct reftable_merged_table *mt)
 322 {
 323         return mt->hash_id;
 324 }
 325
 326 static int reftable_merged_table_seek_void(void *tab,
 327                                            struct reftable_iterator *it,
 328                                            struct reftable_record *rec)
 329 {
 330         return merged_table_seek_record(tab, it, rec);
 331 }
 332
 333 static uint32_t reftable_merged_table_hash_id_void(void *tab)
 334 {
 335         return reftable_merged_table_hash_id(tab);
 336 }
 337
 338 static uint64_t reftable_merged_table_min_update_index_void(void *tab)
 339 {
 340         return reftable_merged_table_min_update_index(tab);
 341 }
 342
 343 static uint64_t reftable_merged_table_max_update_index_void(void *tab)
 344 {
 345         return reftable_merged_table_max_update_index(tab);
 346 }
 347
 348 static struct reftable_table_vtable merged_table_vtable = {
 349         .seek_record = reftable_merged_table_seek_void,
 350         .hash_id = reftable_merged_table_hash_id_void,
 351         .min_update_index = reftable_merged_table_min_update_index_void,
 352         .max_update_index = reftable_merged_table_max_update_index_void,
 353 };
 354
 355 void reftable_table_from_merged_table(struct reftable_table *tab,
 356                                       struct reftable_merged_table *merged)
 357 {
 358         assert(!tab->ops);
 359         tab->ops = &merged_table_vtable;
 360         tab->table_arg = merged;
 361 }