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