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