summaryrefslogtreecommitdiffstats
path: root/src/kv/KeyValueDB.h
blob: 9cfb4482706c967190e9539afc4369b30c184dfe (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
#ifndef KEY_VALUE_DB_H
#define KEY_VALUE_DB_H

#include "include/buffer.h"
#include <ostream>
#include <set>
#include <map>
#include <optional>
#include <string>
#include <boost/scoped_ptr.hpp>
#include "include/encoding.h"
#include "common/Formatter.h"
#include "common/perf_counters.h"
#include "common/PriorityCache.h"

/**
 * Defines virtual interface to be implemented by key value store
 *
 * Kyoto Cabinet should implement this
 */
class KeyValueDB {
public:
  class TransactionImpl {
  public:
    /// Set Keys
    void set(
      const std::string &prefix,                      ///< [in] Prefix for keys, or CF name
      const std::map<std::string, ceph::buffer::list> &to_set ///< [in] keys/values to set
    ) {
      for (auto it = to_set.cbegin(); it != to_set.cend(); ++it)
	set(prefix, it->first, it->second);
    }

    /// Set Keys (via encoded ceph::buffer::list)
    void set(
      const std::string &prefix,      ///< [in] prefix, or CF name
      ceph::buffer::list& to_set_bl           ///< [in] encoded key/values to set
      ) {
      using ceph::decode;
      auto p = std::cbegin(to_set_bl);
      uint32_t num;
      decode(num, p);
      while (num--) {
	std::string key;
	ceph::buffer::list value;
	decode(key, p);
	decode(value, p);
	set(prefix, key, value);
      }
    }

    /// Set Key
    virtual void set(
      const std::string &prefix,      ///< [in] Prefix or CF for the key
      const std::string &k,	      ///< [in] Key to set
      const ceph::buffer::list &bl            ///< [in] Value to set
      ) = 0;
    virtual void set(
      const std::string &prefix,
      const char *k,
      size_t keylen,
      const ceph::buffer::list& bl) {
      set(prefix, std::string(k, keylen), bl);
    }

    /// Removes Keys (via encoded ceph::buffer::list)
    void rmkeys(
      const std::string &prefix,     ///< [in] Prefix or CF to search for
      ceph::buffer::list &keys_bl            ///< [in] Keys to remove
    ) {
      using ceph::decode;
      auto p = std::cbegin(keys_bl);
      uint32_t num;
      decode(num, p);
      while (num--) {
	std::string key;
	decode(key, p);
	rmkey(prefix, key);
      }
    }

    /// Removes Keys
    void rmkeys(
      const std::string &prefix,        ///< [in] Prefix/CF to search for
      const std::set<std::string> &keys ///< [in] Keys to remove
    ) {
      for (auto it = keys.cbegin(); it != keys.cend(); ++it)
	rmkey(prefix, *it);
    }

    /// Remove Key
    virtual void rmkey(
      const std::string &prefix,       ///< [in] Prefix/CF to search for
      const std::string &k	       ///< [in] Key to remove
      ) = 0;
    virtual void rmkey(
      const std::string &prefix,   ///< [in] Prefix to search for
      const char *k,	      ///< [in] Key to remove
      size_t keylen
      ) {
      rmkey(prefix, std::string(k, keylen));
    }

    /// Remove Single Key which exists and was not overwritten.
    /// This API is only related to performance optimization, and should only be 
    /// re-implemented by log-insert-merge tree based keyvalue stores(such as RocksDB). 
    /// If a key is overwritten (by calling set multiple times), then the result
    /// of calling rm_single_key on this key is undefined.
    virtual void rm_single_key(
      const std::string &prefix,      ///< [in] Prefix/CF to search for
      const std::string &k	      ///< [in] Key to remove
      ) { return rmkey(prefix, k);}

    /// Removes keys beginning with prefix
    virtual void rmkeys_by_prefix(
      const std::string &prefix       ///< [in] Prefix/CF by which to remove keys
      ) = 0;

    virtual void rm_range_keys(
      const std::string &prefix,    ///< [in] Prefix by which to remove keys
      const std::string &start,     ///< [in] The start bound of remove keys
      const std::string &end        ///< [in] The start bound of remove keys
      ) = 0;

    /// Merge value into key
    virtual void merge(
      const std::string &prefix,   ///< [in] Prefix/CF ==> MUST match some established merge operator
      const std::string &key,      ///< [in] Key to be merged
      const ceph::buffer::list  &value     ///< [in] value to be merged into key
    ) { ceph_abort_msg("Not implemented"); }

    virtual ~TransactionImpl() {}
  };
  typedef std::shared_ptr< TransactionImpl > Transaction;

  /// create a new instance
  static KeyValueDB *create(CephContext *cct, const std::string& type,
			    const std::string& dir,
			    std::map<std::string,std::string> options = {},
			    void *p = NULL);

  /// test whether we can successfully initialize; may have side effects (e.g., create)
  static int test_init(const std::string& type, const std::string& dir);
  virtual int init(std::string option_str="") = 0;
  virtual int open(std::ostream &out, const std::string& cfs="") = 0;
  // std::vector cfs contains column families to be created when db is created.
  virtual int create_and_open(std::ostream &out, const std::string& cfs="") = 0;

  virtual int open_read_only(std::ostream &out, const std::string& cfs="") {
    return -ENOTSUP;
  }

  virtual void close() { }

  /// Try to repair K/V database. rocksdb requires that database must be not opened.
  virtual int repair(std::ostream &out) { return 0; }

  virtual Transaction get_transaction() = 0;
  virtual int submit_transaction(Transaction) = 0;
  virtual int submit_transaction_sync(Transaction t) {
    return submit_transaction(t);
  }

  /// Retrieve Keys
  virtual int get(
    const std::string &prefix,               ///< [in] Prefix/CF for key
    const std::set<std::string> &key,        ///< [in] Key to retrieve
    std::map<std::string, ceph::buffer::list> *out   ///< [out] Key value retrieved
    ) = 0;
  virtual int get(const std::string &prefix, ///< [in] prefix or CF name
		  const std::string &key,    ///< [in] key
		  ceph::buffer::list *value) {       ///< [out] value
    std::set<std::string> ks;
    ks.insert(key);
    std::map<std::string,ceph::buffer::list> om;
    int r = get(prefix, ks, &om);
    if (om.find(key) != om.end()) {
      *value = std::move(om[key]);
    } else {
      *value = ceph::buffer::list();
      r = -ENOENT;
    }
    return r;
  }
  virtual int get(const std::string &prefix,
		  const char *key, size_t keylen,
		  ceph::buffer::list *value) {
    return get(prefix, std::string(key, keylen), value);
  }

  // This superclass is used both by kv iterators *and* by the ObjectMap
  // omap iterator.  The class hierarchies are unfortunately tied together
  // by the legacy DBOjectMap implementation :(.
  class SimplestIteratorImpl {
  public:
    virtual int seek_to_first() = 0;
    virtual int upper_bound(const std::string &after) = 0;
    virtual int lower_bound(const std::string &to) = 0;
    virtual bool valid() = 0;
    virtual int next() = 0;
    virtual std::string key() = 0;
    virtual std::string tail_key() {
      return "";
    }
    virtual ceph::buffer::list value() = 0;
    virtual int status() = 0;
    virtual ~SimplestIteratorImpl() {}
  };

  class IteratorImpl : public SimplestIteratorImpl {
  public:
    virtual ~IteratorImpl() {}
    virtual int seek_to_last() = 0;
    virtual int prev() = 0;
    virtual std::pair<std::string, std::string> raw_key() = 0;
    virtual ceph::buffer::ptr value_as_ptr() {
      ceph::buffer::list bl = value();
      if (bl.length() == 1) {
        return *bl.buffers().begin();
      } else if (bl.length() == 0) {
        return ceph::buffer::ptr();
      } else {
	ceph_abort();
      }
    }
  };
  typedef std::shared_ptr< IteratorImpl > Iterator;

  // This is the low-level iterator implemented by the underlying KV store.
  class WholeSpaceIteratorImpl {
  public:
    virtual int seek_to_first() = 0;
    virtual int seek_to_first(const std::string &prefix) = 0;
    virtual int seek_to_last() = 0;
    virtual int seek_to_last(const std::string &prefix) = 0;
    virtual int upper_bound(const std::string &prefix, const std::string &after) = 0;
    virtual int lower_bound(const std::string &prefix, const std::string &to) = 0;
    virtual bool valid() = 0;
    virtual int next() = 0;
    virtual int prev() = 0;
    virtual std::string key() = 0;
    virtual std::pair<std::string,std::string> raw_key() = 0;
    virtual bool raw_key_is_prefixed(const std::string &prefix) = 0;
    virtual ceph::buffer::list value() = 0;
    virtual ceph::buffer::ptr value_as_ptr() {
      ceph::buffer::list bl = value();
      if (bl.length()) {
        return *bl.buffers().begin();
      } else {
        return ceph::buffer::ptr();
      }
    }
    virtual int status() = 0;
    virtual size_t key_size() {
      return 0;
    }
    virtual size_t value_size() {
      return 0;
    }
    virtual ~WholeSpaceIteratorImpl() { }
  };
  typedef std::shared_ptr< WholeSpaceIteratorImpl > WholeSpaceIterator;

private:
  // This class filters a WholeSpaceIterator by a prefix.
  // Performs as a dummy wrapper over WholeSpaceIterator
  // if prefix is empty
  class PrefixIteratorImpl : public IteratorImpl {
    const std::string prefix;
    WholeSpaceIterator generic_iter;
  public:
    PrefixIteratorImpl(const std::string &prefix, WholeSpaceIterator iter) :
      prefix(prefix), generic_iter(iter) { }
    ~PrefixIteratorImpl() override { }

    int seek_to_first() override {
      return prefix.empty() ?
	generic_iter->seek_to_first() :
	generic_iter->seek_to_first(prefix);
    }
    int seek_to_last() override {
      return prefix.empty() ?
	generic_iter->seek_to_last() :
	generic_iter->seek_to_last(prefix);
    }
    int upper_bound(const std::string &after) override {
      return generic_iter->upper_bound(prefix, after);
    }
    int lower_bound(const std::string &to) override {
      return generic_iter->lower_bound(prefix, to);
    }
    bool valid() override {
      if (!generic_iter->valid())
	return false;
      if (prefix.empty())
        return true;
      return prefix.empty() ?
	      true :
	      generic_iter->raw_key_is_prefixed(prefix);
    }
    int next() override {
      return generic_iter->next();
    }
    int prev() override {
      return generic_iter->prev();
    }
    std::string key() override {
      return generic_iter->key();
    }
    std::pair<std::string, std::string> raw_key() override {
      return generic_iter->raw_key();
    }
    ceph::buffer::list value() override {
      return generic_iter->value();
    }
    ceph::buffer::ptr value_as_ptr() override {
      return generic_iter->value_as_ptr();
    }
    int status() override {
      return generic_iter->status();
    }
  };
protected:
  Iterator make_iterator(const std::string &prefix, WholeSpaceIterator w_iter) {
    return std::make_shared<PrefixIteratorImpl>(
      prefix,
      w_iter);
  }
public:
  typedef uint32_t IteratorOpts;
  static const uint32_t ITERATOR_NOCACHE = 1;

  struct IteratorBounds {
    std::optional<std::string> lower_bound;
    std::optional<std::string> upper_bound;
  };

  virtual WholeSpaceIterator get_wholespace_iterator(IteratorOpts opts = 0) = 0;
  virtual Iterator get_iterator(const std::string &prefix, IteratorOpts opts = 0, IteratorBounds bounds = IteratorBounds()) {
    return make_iterator(prefix,
      get_wholespace_iterator(opts));
  }

  virtual uint64_t get_estimated_size(std::map<std::string,uint64_t> &extra) = 0;
  virtual int get_statfs(struct store_statfs_t *buf) {
    return -EOPNOTSUPP;
  }

  virtual int set_cache_size(uint64_t) {
    return -EOPNOTSUPP;
  }

  virtual int set_cache_high_pri_pool_ratio(double ratio) {
    return -EOPNOTSUPP;
  }

  virtual int64_t get_cache_usage() const {
    return -EOPNOTSUPP;
  }

  virtual int64_t get_cache_usage(std::string prefix) const {
    return -EOPNOTSUPP;
  }

  virtual std::shared_ptr<PriorityCache::PriCache> get_priority_cache() const {
    return nullptr;
  }

  virtual std::shared_ptr<PriorityCache::PriCache> get_priority_cache(std::string prefix) const {
    return nullptr;
  }



  virtual ~KeyValueDB() {}

  /// estimate space utilization for a prefix (in bytes)
  virtual int64_t estimate_prefix_size(const std::string& prefix,
				       const std::string& key_prefix) {
    return 0;
  }

  /// compact the underlying store
  virtual void compact() {}

  /// compact the underlying store in async mode
  virtual void compact_async() {}

  /// compact db for all keys with a given prefix
  virtual void compact_prefix(const std::string& prefix) {}
  /// compact db for all keys with a given prefix, async
  virtual void compact_prefix_async(const std::string& prefix) {}
  virtual void compact_range(const std::string& prefix,
			     const std::string& start, const std::string& end) {}
  virtual void compact_range_async(const std::string& prefix,
				   const std::string& start, const std::string& end) {}

  // See RocksDB merge operator definition, we support the basic
  // associative merge only right now.
  class MergeOperator {
    public:
    /// Merge into a key that doesn't exist
    virtual void merge_nonexistent(
      const char *rdata, size_t rlen,
      std::string *new_value) = 0;
    /// Merge into a key that does exist
    virtual void merge(
      const char *ldata, size_t llen,
      const char *rdata, size_t rlen,
      std::string *new_value) = 0;
    /// We use each operator name and each prefix to construct the overall RocksDB operator name for consistency check at open time.
    virtual const char *name() const = 0;

    virtual ~MergeOperator() {}
  };

  /// Setup one or more operators, this needs to be done BEFORE the DB is opened.
  virtual int set_merge_operator(const std::string& prefix,
				 std::shared_ptr<MergeOperator> mop) {
    return -EOPNOTSUPP;
  }

  virtual void get_statistics(ceph::Formatter *f) {
    return;
  }

  /**
   * Return your perf counters if you have any.  Subclasses are not
   * required to implement this, and callers must respect a null return
   * value.
   */
  virtual PerfCounters *get_perf_counters() {
    return nullptr;
  }

  /**
   * Access implementation specific integral property corresponding
   * to passed property and prefic.
   * Return value is true if property is valid for prefix, populates out.
   */
  virtual bool get_property(
    const std::string &property,
    uint64_t *out) {
    return false;
  }
protected:
  /// List of matching prefixes/ColumnFamilies and merge operators
  std::vector<std::pair<std::string,
			std::shared_ptr<MergeOperator> > > merge_ops;

};

#endif