2 Copyright 2020 Google LLC
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
11 #include "constants.h"
17 #include "reftable-merged.h"
18 #include "reftable-error.h"
21 static int merged_iter_init(struct merged_iter
*mi
)
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
);
32 reftable_iterator_destroy(&mi
->stack
[i
]);
33 reftable_record_release(&rec
);
39 merged_iter_pqueue_add(&mi
->pq
, &e
);
46 static void merged_iter_close(void *p
)
48 struct merged_iter
*mi
= p
;
50 merged_iter_pqueue_release(&mi
->pq
);
51 for (i
= 0; i
< mi
->stack_len
; i
++) {
52 reftable_iterator_destroy(&mi
->stack
[i
]);
54 reftable_free(mi
->stack
);
55 strbuf_release(&mi
->key
);
56 strbuf_release(&mi
->entry_key
);
59 static int merged_iter_advance_nonnull_subiter(struct merged_iter
*mi
,
63 .rec
= reftable_new_record(mi
->typ
),
66 int err
= iterator_next(&mi
->stack
[idx
], &e
.rec
);
71 reftable_iterator_destroy(&mi
->stack
[idx
]);
72 reftable_record_release(&e
.rec
);
76 merged_iter_pqueue_add(&mi
->pq
, &e
);
80 static int merged_iter_advance_subiter(struct merged_iter
*mi
, size_t idx
)
82 if (iterator_is_null(&mi
->stack
[idx
]))
84 return merged_iter_advance_nonnull_subiter(mi
, idx
);
87 static int merged_iter_next_entry(struct merged_iter
*mi
,
88 struct reftable_record
*rec
)
90 struct pq_entry entry
= { 0 };
93 if (merged_iter_pqueue_is_empty(mi
->pq
))
96 entry
= merged_iter_pqueue_remove(&mi
->pq
);
97 err
= merged_iter_advance_subiter(mi
, entry
.index
);
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.
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
);
114 reftable_record_key(&top
.rec
, &mi
->key
);
116 cmp
= strbuf_cmp(&mi
->key
, &mi
->entry_key
);
120 merged_iter_pqueue_remove(&mi
->pq
);
121 err
= merged_iter_advance_subiter(mi
, top
.index
);
124 reftable_record_release(&top
.rec
);
127 reftable_record_copy_from(rec
, &entry
.rec
, hash_size(mi
->hash_id
));
130 reftable_record_release(&entry
.rec
);
131 strbuf_release(&mi
->entry_key
);
132 strbuf_release(&mi
->key
);
136 static int merged_iter_next(struct merged_iter
*mi
, struct reftable_record
*rec
)
139 int err
= merged_iter_next_entry(mi
, rec
);
140 if (err
== 0 && mi
->suppress_deletions
&&
141 reftable_record_is_deletion(rec
)) {
149 static int merged_iter_next_void(void *p
, struct reftable_record
*rec
)
151 struct merged_iter
*mi
= p
;
152 if (merged_iter_pqueue_is_empty(mi
->pq
))
155 return merged_iter_next(mi
, rec
);
158 static struct reftable_iterator_vtable merged_iter_vtable
= {
159 .next
= &merged_iter_next_void
,
160 .close
= &merged_iter_close
,
163 static void iterator_from_merged_iter(struct reftable_iterator
*it
,
164 struct merged_iter
*mi
)
168 it
->ops
= &merged_iter_vtable
;
171 int reftable_new_merged_table(struct reftable_merged_table
**dest
,
172 struct reftable_table
*stack
, int n
,
175 struct reftable_merged_table
*m
= NULL
;
176 uint64_t last_max
= 0;
177 uint64_t first_min
= 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
]);
183 if (reftable_table_hash_id(&stack
[i
]) != hash_id
) {
184 return REFTABLE_FORMAT_ERROR
;
186 if (i
== 0 || min
< first_min
) {
189 if (i
== 0 || max
> last_max
) {
194 m
= reftable_calloc(sizeof(struct reftable_merged_table
));
199 m
->hash_id
= hash_id
;
204 /* clears the list of subtable, without affecting the readers themselves. */
205 void merged_table_release(struct reftable_merged_table
*mt
)
207 FREE_AND_NULL(mt
->stack
);
211 void reftable_merged_table_free(struct reftable_merged_table
*mt
)
216 merged_table_release(mt
);
221 reftable_merged_table_max_update_index(struct reftable_merged_table
*mt
)
227 reftable_merged_table_min_update_index(struct reftable_merged_table
*mt
)
232 static int reftable_table_seek_record(struct reftable_table
*tab
,
233 struct reftable_iterator
*it
,
234 struct reftable_record
*rec
)
236 return tab
->ops
->seek_record(tab
->table_arg
, it
, rec
);
239 static int merged_table_seek_record(struct reftable_merged_table
*mt
,
240 struct reftable_iterator
*it
,
241 struct reftable_record
*rec
)
243 struct reftable_iterator
*iters
= reftable_calloc(
244 sizeof(struct reftable_iterator
) * mt
->stack_len
);
245 struct merged_iter merged
= {
247 .typ
= reftable_record_type(rec
),
248 .hash_id
= mt
->hash_id
,
249 .suppress_deletions
= mt
->suppress_deletions
,
251 .entry_key
= STRBUF_INIT
,
256 for (i
= 0; i
< mt
->stack_len
&& err
== 0; i
++) {
257 int e
= reftable_table_seek_record(&mt
->stack
[i
], &iters
[n
],
268 for (i
= 0; i
< n
; i
++) {
269 reftable_iterator_destroy(&iters
[i
]);
271 reftable_free(iters
);
275 merged
.stack_len
= n
;
276 err
= merged_iter_init(&merged
);
278 merged_iter_close(&merged
);
281 struct merged_iter
*p
=
282 reftable_malloc(sizeof(struct merged_iter
));
284 iterator_from_merged_iter(it
, p
);
289 int reftable_merged_table_seek_ref(struct reftable_merged_table
*mt
,
290 struct reftable_iterator
*it
,
293 struct reftable_record rec
= {
294 .type
= BLOCK_TYPE_REF
,
296 .refname
= (char *)name
,
299 return merged_table_seek_record(mt
, it
, &rec
);
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
)
306 struct reftable_record rec
= { .type
= BLOCK_TYPE_LOG
,
308 .refname
= (char *)name
,
309 .update_index
= update_index
,
311 return merged_table_seek_record(mt
, it
, &rec
);
314 int reftable_merged_table_seek_log(struct reftable_merged_table
*mt
,
315 struct reftable_iterator
*it
,
318 uint64_t max
= ~((uint64_t)0);
319 return reftable_merged_table_seek_log_at(mt
, it
, name
, max
);
322 uint32_t reftable_merged_table_hash_id(struct reftable_merged_table
*mt
)
327 static int reftable_merged_table_seek_void(void *tab
,
328 struct reftable_iterator
*it
,
329 struct reftable_record
*rec
)
331 return merged_table_seek_record(tab
, it
, rec
);
334 static uint32_t reftable_merged_table_hash_id_void(void *tab
)
336 return reftable_merged_table_hash_id(tab
);
339 static uint64_t reftable_merged_table_min_update_index_void(void *tab
)
341 return reftable_merged_table_min_update_index(tab
);
344 static uint64_t reftable_merged_table_max_update_index_void(void *tab
)
346 return reftable_merged_table_max_update_index(tab
);
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
,
356 void reftable_table_from_merged_table(struct reftable_table
*tab
,
357 struct reftable_merged_table
*merged
)
360 tab
->ops
= &merged_table_vtable
;
361 tab
->table_arg
= merged
;