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 }
  56
  57 static int merged_iter_advance_nonnull_subiter(struct merged_iter *mi,
  58                                                size_t idx)
  59 {
  60         struct pq_entry e = {
  61                 .rec = reftable_new_record(mi->typ),
  62                 .index = idx,
  63         };
  64         int err = iterator_next(&mi->stack[idx], &e.rec);
  65         if (err < 0)
  66                 return err;
  67
  68         if (err > 0) {
  69                 reftable_iterator_destroy(&mi->stack[idx]);
  70                 reftable_record_release(&e.rec);
  71                 return 0;
  72         }
  73
  74         merged_iter_pqueue_add(&mi->pq, &e);
  75         return 0;
  76 }
  77
  78 static int merged_iter_advance_subiter(struct merged_iter *mi, size_t idx)
  79 {
  80         if (iterator_is_null(&mi->stack[idx]))
  81                 return 0;
  82         return merged_iter_advance_nonnull_subiter(mi, idx);
  83 }
  84
  85 static int merged_iter_next_entry(struct merged_iter *mi,
  86                                   struct reftable_record *rec)
  87 {
  88         struct strbuf entry_key = STRBUF_INIT;
  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, &entry_key);
 109         while (!merged_iter_pqueue_is_empty(mi->pq)) {
 110                 struct pq_entry top = merged_iter_pqueue_top(mi->pq);
 111                 struct strbuf k = STRBUF_INIT;
 112                 int err = 0, cmp = 0;
 113
 114                 reftable_record_key(&top.rec, &k);
 115
 116                 cmp = strbuf_cmp(&k, &entry_key);
 117                 strbuf_release(&k);
 118
 119                 if (cmp > 0) {
 120                         break;
 121                 }
 122
 123                 merged_iter_pqueue_remove(&mi->pq);
 124                 err = merged_iter_advance_subiter(mi, top.index);
 125                 if (err < 0) {
 126                         return err;
 127                 }
 128                 reftable_record_release(&top.rec);
 129         }
 130
 131         reftable_record_copy_from(rec, &entry.rec, hash_size(mi->hash_id));
 132         reftable_record_release(&entry.rec);
 133         strbuf_release(&entry_key);
 134         return 0;
 135 }
 136
 137 static int merged_iter_next(struct merged_iter *mi, struct reftable_record *rec)
 138 {
 139         while (1) {
 140                 int err = merged_iter_next_entry(mi, rec);
 141                 if (err == 0 && mi->suppress_deletions &&
 142                     reftable_record_is_deletion(rec)) {
 143                         continue;
 144                 }
 145
 146                 return err;
 147         }
 148 }
 149
 150 static int merged_iter_next_void(void *p, struct reftable_record *rec)
 151 {
 152         struct merged_iter *mi = p;
 153         if (merged_iter_pqueue_is_empty(mi->pq))
 154                 return 1;
 155
 156         return merged_iter_next(mi, rec);
 157 }
 158
 159 static struct reftable_iterator_vtable merged_iter_vtable = {
 160         .next = &merged_iter_next_void,
 161         .close = &merged_iter_close,
 162 };
 163
 164 static void iterator_from_merged_iter(struct reftable_iterator *it,
 165                                       struct merged_iter *mi)
 166 {
 167         assert(!it->ops);
 168         it->iter_arg = mi;
 169         it->ops = &merged_iter_vtable;
 170 }
 171
 172 int reftable_new_merged_table(struct reftable_merged_table **dest,
 173                               struct reftable_table *stack, int n,
 174                               uint32_t hash_id)
 175 {
 176         struct reftable_merged_table *m = NULL;
 177         uint64_t last_max = 0;
 178         uint64_t first_min = 0;
 179         int i = 0;
 180         for (i = 0; i < n; i++) {
 181                 uint64_t min = reftable_table_min_update_index(&stack[i]);
 182                 uint64_t max = reftable_table_max_update_index(&stack[i]);
 183
 184                 if (reftable_table_hash_id(&stack[i]) != hash_id) {
 185                         return REFTABLE_FORMAT_ERROR;
 186                 }
 187                 if (i == 0 || min < first_min) {
 188                         first_min = min;
 189                 }
 190                 if (i == 0 || max > last_max) {
 191                         last_max = max;
 192                 }
 193         }
 194
 195         m = reftable_calloc(sizeof(struct reftable_merged_table));
 196         m->stack = stack;
 197         m->stack_len = n;
 198         m->min = first_min;
 199         m->max = last_max;
 200         m->hash_id = hash_id;
 201         *dest = m;
 202         return 0;
 203 }
 204
 205 /* clears the list of subtable, without affecting the readers themselves. */
 206 void merged_table_release(struct reftable_merged_table *mt)
 207 {
 208         FREE_AND_NULL(mt->stack);
 209         mt->stack_len = 0;
 210 }
 211
 212 void reftable_merged_table_free(struct reftable_merged_table *mt)
 213 {
 214         if (!mt) {
 215                 return;
 216         }
 217         merged_table_release(mt);
 218         reftable_free(mt);
 219 }
 220
 221 uint64_t
 222 reftable_merged_table_max_update_index(struct reftable_merged_table *mt)
 223 {
 224         return mt->max;
 225 }
 226
 227 uint64_t
 228 reftable_merged_table_min_update_index(struct reftable_merged_table *mt)
 229 {
 230         return mt->min;
 231 }
 232
 233 static int reftable_table_seek_record(struct reftable_table *tab,
 234                                       struct reftable_iterator *it,
 235                                       struct reftable_record *rec)
 236 {
 237         return tab->ops->seek_record(tab->table_arg, it, rec);
 238 }
 239
 240 static int merged_table_seek_record(struct reftable_merged_table *mt,
 241                                     struct reftable_iterator *it,
 242                                     struct reftable_record *rec)
 243 {
 244         struct reftable_iterator *iters = reftable_calloc(
 245                 sizeof(struct reftable_iterator) * mt->stack_len);
 246         struct merged_iter merged = {
 247                 .stack = iters,
 248                 .typ = reftable_record_type(rec),
 249                 .hash_id = mt->hash_id,
 250                 .suppress_deletions = mt->suppress_deletions,
 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 }