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