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         for (size_t i = 0; i < mi->stack_len; i++) {
  23                 struct pq_entry e = {
  24                         .index = i,
  25                 };
  26                 int err;
  27
  28                 reftable_record_init(&e.rec, mi->typ);
  29                 err = iterator_next(&mi->stack[i], &e.rec);
  30                 if (err < 0)
  31                         return err;
  32                 if (err > 0) {
  33                         reftable_iterator_destroy(&mi->stack[i]);
  34                         reftable_record_release(&e.rec);
  35                         continue;
  36                 }
  37
  38                 merged_iter_pqueue_add(&mi->pq, &e);
  39         }
  40
  41         return 0;
  42 }
  43
  44 static void merged_iter_close(void *p)
  45 {
  46         struct merged_iter *mi = p;
  47
  48         merged_iter_pqueue_release(&mi->pq);
  49         for (size_t i = 0; i < mi->stack_len; i++)
  50                 reftable_iterator_destroy(&mi->stack[i]);
  51         reftable_free(mi->stack);
  52         strbuf_release(&mi->key);
  53         strbuf_release(&mi->entry_key);
  54 }
  55
  56 static int merged_iter_advance_nonnull_subiter(struct merged_iter *mi,
  57                                                size_t idx)
  58 {
  59         struct pq_entry e = {
  60                 .index = idx,
  61         };
  62         int err;
  63
  64         reftable_record_init(&e.rec, mi->typ);
  65         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, size_t 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
 178         for (size_t 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         REFTABLE_CALLOC_ARRAY(m, 1);
 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 merged_iter merged = {
 243                 .typ = reftable_record_type(rec),
 244                 .hash_id = mt->hash_id,
 245                 .suppress_deletions = mt->suppress_deletions,
 246                 .key = STRBUF_INIT,
 247                 .entry_key = STRBUF_INIT,
 248         };
 249         struct merged_iter *p;
 250         int err;
 251
 252         REFTABLE_CALLOC_ARRAY(merged.stack, mt->stack_len);
 253         for (size_t i = 0; i < mt->stack_len; i++) {
 254                 err = reftable_table_seek_record(&mt->stack[i],
 255                                                  &merged.stack[merged.stack_len], rec);
 256                 if (err < 0)
 257                         goto out;
 258                 if (!err)
 259                         merged.stack_len++;
 260         }
 261
 262         err = merged_iter_init(&merged);
 263         if (err < 0)
 264                 goto out;
 265
 266         p = reftable_malloc(sizeof(struct merged_iter));
 267         *p = merged;
 268         iterator_from_merged_iter(it, p);
 269
 270 out:
 271         if (err < 0)
 272                 merged_iter_close(&merged);
 273         return err;
 274 }
 275
 276 int reftable_merged_table_seek_ref(struct reftable_merged_table *mt,
 277                                    struct reftable_iterator *it,
 278                                    const char *name)
 279 {
 280         struct reftable_record rec = {
 281                 .type = BLOCK_TYPE_REF,
 282                 .u.ref = {
 283                         .refname = (char *)name,
 284                 },
 285         };
 286         return merged_table_seek_record(mt, it, &rec);
 287 }
 288
 289 int reftable_merged_table_seek_log_at(struct reftable_merged_table *mt,
 290                                       struct reftable_iterator *it,
 291                                       const char *name, uint64_t update_index)
 292 {
 293         struct reftable_record rec = { .type = BLOCK_TYPE_LOG,
 294                                        .u.log = {
 295                                                .refname = (char *)name,
 296                                                .update_index = update_index,
 297                                        } };
 298         return merged_table_seek_record(mt, it, &rec);
 299 }
 300
 301 int reftable_merged_table_seek_log(struct reftable_merged_table *mt,
 302                                    struct reftable_iterator *it,
 303                                    const char *name)
 304 {
 305         uint64_t max = ~((uint64_t)0);
 306         return reftable_merged_table_seek_log_at(mt, it, name, max);
 307 }
 308
 309 uint32_t reftable_merged_table_hash_id(struct reftable_merged_table *mt)
 310 {
 311         return mt->hash_id;
 312 }
 313
 314 static int reftable_merged_table_seek_void(void *tab,
 315                                            struct reftable_iterator *it,
 316                                            struct reftable_record *rec)
 317 {
 318         return merged_table_seek_record(tab, it, rec);
 319 }
 320
 321 static uint32_t reftable_merged_table_hash_id_void(void *tab)
 322 {
 323         return reftable_merged_table_hash_id(tab);
 324 }
 325
 326 static uint64_t reftable_merged_table_min_update_index_void(void *tab)
 327 {
 328         return reftable_merged_table_min_update_index(tab);
 329 }
 330
 331 static uint64_t reftable_merged_table_max_update_index_void(void *tab)
 332 {
 333         return reftable_merged_table_max_update_index(tab);
 334 }
 335
 336 static struct reftable_table_vtable merged_table_vtable = {
 337         .seek_record = reftable_merged_table_seek_void,
 338         .hash_id = reftable_merged_table_hash_id_void,
 339         .min_update_index = reftable_merged_table_min_update_index_void,
 340         .max_update_index = reftable_merged_table_max_update_index_void,
 341 };
 342
 343 void reftable_table_from_merged_table(struct reftable_table *tab,
 344                                       struct reftable_merged_table *merged)
 345 {
 346         assert(!tab->ops);
 347         tab->ops = &merged_table_vtable;
 348         tab->table_arg = merged;
 349 }