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"
16 #include "reftable-merged.h"
17 #include "reftable-error.h"
20 static int merged_iter_init(struct merged_iter
*mi
)
22 for (size_t i
= 0; i
< mi
->stack_len
; i
++) {
28 reftable_record_init(&e
.rec
, mi
->typ
);
29 err
= iterator_next(&mi
->stack
[i
], &e
.rec
);
33 reftable_iterator_destroy(&mi
->stack
[i
]);
34 reftable_record_release(&e
.rec
);
38 merged_iter_pqueue_add(&mi
->pq
, &e
);
44 static void merged_iter_close(void *p
)
46 struct merged_iter
*mi
= p
;
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
);
54 static int merged_iter_advance_nonnull_subiter(struct merged_iter
*mi
,
62 reftable_record_init(&e
.rec
, mi
->typ
);
63 err
= iterator_next(&mi
->stack
[idx
], &e
.rec
);
68 reftable_iterator_destroy(&mi
->stack
[idx
]);
69 reftable_record_release(&e
.rec
);
73 merged_iter_pqueue_add(&mi
->pq
, &e
);
77 static int merged_iter_advance_subiter(struct merged_iter
*mi
, size_t idx
)
79 if (iterator_is_null(&mi
->stack
[idx
]))
81 return merged_iter_advance_nonnull_subiter(mi
, idx
);
84 static int merged_iter_next_entry(struct merged_iter
*mi
,
85 struct reftable_record
*rec
)
87 struct pq_entry entry
= { 0 };
90 if (merged_iter_pqueue_is_empty(mi
->pq
))
93 entry
= merged_iter_pqueue_remove(&mi
->pq
);
94 err
= merged_iter_advance_subiter(mi
, entry
.index
);
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.
106 while (!merged_iter_pqueue_is_empty(mi
->pq
)) {
107 struct pq_entry top
= merged_iter_pqueue_top(mi
->pq
);
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.
115 if (top
.index
== entry
.index
)
118 cmp
= reftable_record_cmp(&top
.rec
, &entry
.rec
);
122 merged_iter_pqueue_remove(&mi
->pq
);
123 err
= merged_iter_advance_subiter(mi
, top
.index
);
126 reftable_record_release(&top
.rec
);
129 reftable_record_release(rec
);
134 reftable_record_release(&entry
.rec
);
138 static int merged_iter_next(struct merged_iter
*mi
, struct reftable_record
*rec
)
141 int err
= merged_iter_next_entry(mi
, rec
);
142 if (err
== 0 && mi
->suppress_deletions
&&
143 reftable_record_is_deletion(rec
)) {
151 static int merged_iter_next_void(void *p
, struct reftable_record
*rec
)
153 struct merged_iter
*mi
= p
;
154 if (merged_iter_pqueue_is_empty(mi
->pq
))
157 return merged_iter_next(mi
, rec
);
160 static struct reftable_iterator_vtable merged_iter_vtable
= {
161 .next
= &merged_iter_next_void
,
162 .close
= &merged_iter_close
,
165 static void iterator_from_merged_iter(struct reftable_iterator
*it
,
166 struct merged_iter
*mi
)
170 it
->ops
= &merged_iter_vtable
;
173 int reftable_new_merged_table(struct reftable_merged_table
**dest
,
174 struct reftable_table
*stack
, size_t n
,
177 struct reftable_merged_table
*m
= NULL
;
178 uint64_t last_max
= 0;
179 uint64_t first_min
= 0;
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
]);
185 if (reftable_table_hash_id(&stack
[i
]) != hash_id
) {
186 return REFTABLE_FORMAT_ERROR
;
188 if (i
== 0 || min
< first_min
) {
191 if (i
== 0 || max
> last_max
) {
196 REFTABLE_CALLOC_ARRAY(m
, 1);
201 m
->hash_id
= hash_id
;
206 /* clears the list of subtable, without affecting the readers themselves. */
207 void merged_table_release(struct reftable_merged_table
*mt
)
209 FREE_AND_NULL(mt
->stack
);
213 void reftable_merged_table_free(struct reftable_merged_table
*mt
)
218 merged_table_release(mt
);
223 reftable_merged_table_max_update_index(struct reftable_merged_table
*mt
)
229 reftable_merged_table_min_update_index(struct reftable_merged_table
*mt
)
234 static int reftable_table_seek_record(struct reftable_table
*tab
,
235 struct reftable_iterator
*it
,
236 struct reftable_record
*rec
)
238 return tab
->ops
->seek_record(tab
->table_arg
, it
, rec
);
241 static int merged_table_seek_record(struct reftable_merged_table
*mt
,
242 struct reftable_iterator
*it
,
243 struct reftable_record
*rec
)
245 struct merged_iter merged
= {
246 .typ
= reftable_record_type(rec
),
247 .hash_id
= mt
->hash_id
,
248 .suppress_deletions
= mt
->suppress_deletions
,
250 struct merged_iter
*p
;
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
);
263 err
= merged_iter_init(&merged
);
267 p
= reftable_malloc(sizeof(struct merged_iter
));
269 iterator_from_merged_iter(it
, p
);
273 merged_iter_close(&merged
);
277 int reftable_merged_table_seek_ref(struct reftable_merged_table
*mt
,
278 struct reftable_iterator
*it
,
281 struct reftable_record rec
= {
282 .type
= BLOCK_TYPE_REF
,
284 .refname
= (char *)name
,
287 return merged_table_seek_record(mt
, it
, &rec
);
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
)
294 struct reftable_record rec
= { .type
= BLOCK_TYPE_LOG
,
296 .refname
= (char *)name
,
297 .update_index
= update_index
,
299 return merged_table_seek_record(mt
, it
, &rec
);
302 int reftable_merged_table_seek_log(struct reftable_merged_table
*mt
,
303 struct reftable_iterator
*it
,
306 uint64_t max
= ~((uint64_t)0);
307 return reftable_merged_table_seek_log_at(mt
, it
, name
, max
);
310 uint32_t reftable_merged_table_hash_id(struct reftable_merged_table
*mt
)
315 static int reftable_merged_table_seek_void(void *tab
,
316 struct reftable_iterator
*it
,
317 struct reftable_record
*rec
)
319 return merged_table_seek_record(tab
, it
, rec
);
322 static uint32_t reftable_merged_table_hash_id_void(void *tab
)
324 return reftable_merged_table_hash_id(tab
);
327 static uint64_t reftable_merged_table_min_update_index_void(void *tab
)
329 return reftable_merged_table_min_update_index(tab
);
332 static uint64_t reftable_merged_table_max_update_index_void(void *tab
)
334 return reftable_merged_table_max_update_index(tab
);
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
,
344 void reftable_table_from_merged_table(struct reftable_table
*tab
,
345 struct reftable_merged_table
*merged
)
348 tab
->ops
= &merged_table_vtable
;
349 tab
->table_arg
= merged
;