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