Class: SlateDb::Database

Inherits:
Object
  • Object
show all
Defined in:
lib/slatedb/database.rb

Overview

rubocop:disable Metrics/ClassLength

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.open(path, url: nil, merge_operator: nil) {|db| ... } ⇒ Database

Open a database at the given path.

Examples:

Open a database

db = SlateDb::Database.open("/tmp/mydb")
db.put("key", "value")
db.close

Open with block (auto-close)

SlateDb::Database.open("/tmp/mydb") do |db|
  db.put("key", "value")
end # automatically closed

Open with S3 backend

db = SlateDb::Database.open("/tmp/mydb", url: "s3://mybucket/path")

Open with a custom merge operator (Proc)

# Custom merge that adds numbers
db = SlateDb::Database.open("/tmp/mydb", merge_operator: ->(key, existing, new_val) {
  existing_num = existing ? existing.to_i : 0
  (existing_num + new_val.to_i).to_s
})
db.merge("counter", "5")
db.merge("counter", "3")
db.get("counter") # => "8"

Parameters:

  • path (String) —

    The path identifier for the database

  • url (String, nil) (defaults to: nil) —

    Optional object store URL (e.g., "s3://bucket/path")

  • merge_operator (Symbol, String, Proc, nil) (defaults to: nil) —

    Optional merge operator. Can be a symbol/string ("string_concat" or "concat") or a Proc/lambda that takes (key, existing_value, new_value) and returns the merged value.

Yields:

  • (db) —

    If a block is given, yields the database and ensures it's closed

Returns:

  • (Database) —

    The opened database (or block result if block given)



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
 
# File 'lib/slatedb/database.rb', line 41

def open(path, url: nil, merge_operator: nil)
  opts = {}

  case merge_operator
  when Symbol, String
    opts[:merge_operator] = merge_operator.to_s
  when Proc
    # Store the proc to prevent GC and pass to Rust
    @_merge_operator_proc = merge_operator
    opts[:merge_operator_proc] = merge_operator
  end

  db = _open(path, url, opts)

  if block_given?
    begin
      yield db
    ensure
      begin
        db.close
      rescue StandardError
        nil
      end
    end
  else
    db
  end
end

Instance Method Details

#batch(await_durable: nil, seqnum: nil) {|batch| ... } ⇒ void

This method returns an undefined value.

Create and write a batch using a block.

Examples:

db.batch do |b|
  b.put("key1", "value1")
  b.put("key2", "value2")
  b.delete("old_key")
end

Parameters:

  • await_durable (Boolean) (defaults to: nil) —

    Whether to wait for durability (default: true)

  • seqnum (Integer, nil) (defaults to: nil) —

    User-supplied sequence number applied to the batch. See #put for semantics. (Requires SlateDB >= 0.13.0)

Yields:

  • (batch) —

    Yields a WriteBatch to the block



382
383
384
385
386
 
# File 'lib/slatedb/database.rb', line 382

def batch(await_durable: nil, seqnum: nil)
  b = WriteBatch.new
  yield b
  write(b, await_durable: await_durable, seqnum: seqnum)
end

#begin_transaction(isolation: nil) {|txn| ... } ⇒ Transaction, Object

Begin a new transaction.

Examples:

Manual transaction management

txn = db.begin_transaction
txn.put("key", "value")
txn.commit

Block-based transaction (auto-commit)

db.transaction do |txn|
  txn.put("key", "value")
  txn.get("other_key")
end # automatically committed

Serializable isolation

db.transaction(isolation: :serializable) do |txn|
  val = txn.get("counter")
  txn.put("counter", (val.to_i + 1).to_s)
end

Parameters:

  • isolation (Symbol, String) (defaults to: nil) —

    Isolation level (:snapshot or :serializable)

Yields:

  • (txn) —

    If a block is given, yields the transaction and auto-commits/rollbacks

Returns:

  • (Transaction, Object) —

    The transaction (or block result if block given)



411
412
413
414
 
# File 'lib/slatedb/database.rb', line 411

def begin_transaction(isolation: nil)
  isolation_str = isolation&.to_s
  _begin_transaction(isolation_str)
end

#create_checkpoint(lifetime: nil, name: nil) ⇒ Hash

Create a checkpoint of the database.

Examples:

Create a named checkpoint

checkpoint = db.create_checkpoint(name: "before-migration")
puts "Checkpoint ID: #{checkpoint[:id]}"

Create a checkpoint with lifetime

checkpoint = db.create_checkpoint(lifetime: 3600_000) # 1 hour

Parameters:

  • lifetime (Integer, nil) (defaults to: nil) —

    Checkpoint lifetime in milliseconds

  • name (String, nil) (defaults to: nil) —

    Optional name for the checkpoint

Returns:

  • (Hash) —

    Hash with :id (UUID string) and :manifest_id (integer)



496
497
498
499
500
501
 
# File 'lib/slatedb/database.rb', line 496

def create_checkpoint(lifetime: nil, name: nil)
  opts = {}
  opts[:lifetime] = lifetime if lifetime
  opts[:name] = name if name
  _create_checkpoint(opts)
end

#delete(key, await_durable: nil, seqnum: nil) ⇒ void

This method returns an undefined value.

Delete a key.

Examples:

Basic delete

db.delete("mykey")

Delete without waiting for durability

db.delete("mykey", await_durable: false)

Parameters:

  • key (String) —

    The key to delete

  • await_durable (Boolean) (defaults to: nil) —

    Whether to wait for durability (default: true)

  • seqnum (Integer, nil) (defaults to: nil) —

    User-supplied sequence number for this write. See #put for semantics. (Requires SlateDB >= 0.13.0)



177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/slatedb/database.rb', line 177

def delete(key, await_durable: nil, seqnum: nil)
  opts = {}
  opts[:await_durable] = await_durable unless await_durable.nil?
  opts[:seqnum] = seqnum if seqnum

  if opts.empty?
    _delete(key)
  else
    _delete_with_options(key, opts)
  end
end

#get(key, durability_filter: nil, dirty: nil, cache_blocks: nil) ⇒ String?

Get a value by key.

Examples:

Basic get

value = db.get("mykey")

Get with options

value = db.get("mykey", durability_filter: "memory", dirty: true)

Parameters:

  • key (String) —

    The key to look up

  • durability_filter (String, nil) (defaults to: nil) —

    Filter by durability level ("remote" or "memory")

  • dirty (Boolean, nil) (defaults to: nil) —

    Whether to include uncommitted data

  • cache_blocks (Boolean, nil) (defaults to: nil) —

    Whether to cache blocks

Returns:

  • (String, nil) —

    The value, or nil if not found



85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/slatedb/database.rb', line 85

def get(key, durability_filter: nil, dirty: nil, cache_blocks: nil)
  opts = {}
  opts[:durability_filter] = durability_filter.to_s if durability_filter
  opts[:dirty] = dirty unless dirty.nil?
  opts[:cache_blocks] = cache_blocks unless cache_blocks.nil?

  if opts.empty?
    _get(key)
  else
    _get_with_options(key, opts)
  end
end

#get_key_value(key, durability_filter: nil, dirty: nil, cache_blocks: nil) ⇒ Hash? Also known as: get_entry

Get a key-value pair with SlateDB metadata.

Examples:

Inspect metadata

entry = db.get_key_value("mykey")
entry[:value] # => "myvalue"
entry[:seq]   # => sequence number

Parameters:

  • key (String) —

    The key to look up

  • durability_filter (String, nil) (defaults to: nil) —

    Filter by durability level ("remote" or "memory")

  • dirty (Boolean, nil) (defaults to: nil) —

    Whether to include uncommitted data

  • cache_blocks (Boolean, nil) (defaults to: nil) —

    Whether to cache blocks

Returns:

  • (Hash, nil) —

    A hash with :key, :value, :seq, :create_ts, and :expire_ts, or nil if not found



111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/slatedb/database.rb', line 111

def get_key_value(key, durability_filter: nil, dirty: nil, cache_blocks: nil)
  opts = {}
  opts[:durability_filter] = durability_filter.to_s if durability_filter
  opts[:dirty] = dirty unless dirty.nil?
  opts[:cache_blocks] = cache_blocks unless cache_blocks.nil?

  if opts.empty?
    _get_key_value(key)
  else
    _get_key_value_with_options(key, opts)
  end
end

#merge(key, value, ttl: nil, await_durable: nil, seqnum: nil) ⇒ void

This method returns an undefined value.

Merge a value into the database.

Examples:

Merge with string concatenation operator

db = SlateDb::Database.open("/tmp/mydb", merge_operator: :string_concat)
db.merge("key", "part1")
db.merge("key", "part2")

Parameters:

  • key (String) —

    The key to merge into

  • value (String) —

    The merge operand to apply

  • ttl (Integer, nil) (defaults to: nil) —

    Time-to-live in milliseconds

  • await_durable (Boolean) (defaults to: nil) —

    Whether to wait for durability (default: true)

  • seqnum (Integer, nil) (defaults to: nil) —

    User-supplied sequence number for this write. See #put for semantics. (Requires SlateDB >= 0.13.0)



354
355
356
357
358
359
360
361
362
363
364
365
 
# File 'lib/slatedb/database.rb', line 354

def merge(key, value, ttl: nil, await_durable: nil, seqnum: nil)
  opts = {}
  opts[:ttl] = ttl if ttl
  opts[:await_durable] = await_durable unless await_durable.nil?
  opts[:seqnum] = seqnum if seqnum

  if opts.empty?
    _merge(key, value)
  else
    _merge_with_options(key, value, opts)
  end
end

#metrics ⇒ Metrics

Get database metrics registry.

Returns:



506
507
508
 
# File 'lib/slatedb/database.rb', line 506

def metrics
  _metrics
end

#put(key, value, ttl: nil, await_durable: nil, seqnum: nil) ⇒ void

This method returns an undefined value.

Store a key-value pair.

Examples:

Basic put

db.put("mykey", "myvalue")

Put with TTL

db.put("mykey", "myvalue", ttl: 60_000) # expires in 60 seconds

Put without waiting for durability

db.put("mykey", "myvalue", await_durable: false)

Put with an explicit sequence number

db.put("mykey", "myvalue", seqnum: 42)

Parameters:

  • key (String) —

    The key to store

  • value (String) —

    The value to store

  • ttl (Integer, nil) (defaults to: nil) —

    Time-to-live in milliseconds

  • await_durable (Boolean) (defaults to: nil) —

    Whether to wait for durability (default: true)

  • seqnum (Integer, nil) (defaults to: nil) —

    User-supplied sequence number for this write. When provided (and non-zero), it is used instead of the internally generated sequence number. It must be strictly greater than the current maximum sequence number or the write fails. (Requires SlateDB >= 0.13.0)



150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/slatedb/database.rb', line 150

def put(key, value, ttl: nil, await_durable: nil, seqnum: nil)
  opts = {}
  opts[:ttl] = ttl if ttl
  opts[:await_durable] = await_durable unless await_durable.nil?
  opts[:seqnum] = seqnum if seqnum

  if opts.empty?
    _put(key, value)
  else
    _put_with_options(key, value, opts)
  end
end

#scan(start_key, end_key = nil, durability_filter: nil, dirty: nil, read_ahead_bytes: nil, cache_blocks: nil, max_fetch_tasks: nil, order: nil) ⇒ Iterator

Scan a range of keys.

Examples:

Basic scan

iter = db.scan("a")
while entry = iter.next_entry
  key, value = entry
  puts "#{key}: #{value}"
end

Scan with range

iter = db.scan("a", "z")

Scan with block

db.scan("user:") do |key, value|
  puts "#{key}: #{value}"
end

Parameters:

  • start_key (String) —

    The start key (inclusive)

  • end_key (String, nil) (defaults to: nil) —

    The end key (exclusive). If nil, scans to end.

  • durability_filter (String, nil) (defaults to: nil) —

    Filter by durability level ("remote" or "memory")

  • dirty (Boolean, nil) (defaults to: nil) —

    Whether to include uncommitted data

  • read_ahead_bytes (Integer, nil) (defaults to: nil) —

    Number of bytes to read ahead

  • cache_blocks (Boolean, nil) (defaults to: nil) —

    Whether to cache blocks

  • max_fetch_tasks (Integer, nil) (defaults to: nil) —

    Maximum number of fetch tasks

  • order (Symbol, String, nil) (defaults to: nil) —

    Iteration order (:asc/:ascending or :desc/:descending)

Returns:

  • (Iterator) —

    An iterator over key-value pairs



216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/slatedb/database.rb', line 216

def scan(start_key, end_key = nil, durability_filter: nil, dirty: nil,
         read_ahead_bytes: nil, cache_blocks: nil, max_fetch_tasks: nil, order: nil, &)
  opts = scan_options(
    durability_filter: durability_filter,
    dirty: dirty,
    read_ahead_bytes: read_ahead_bytes,
    cache_blocks: cache_blocks,
    max_fetch_tasks: max_fetch_tasks,
    order: order
  )

  iter = if opts.empty?
           _scan(start_key, end_key)
         else
           _scan_with_options(start_key, end_key, opts)
         end

  if block_given?
    iter.each(&)
  else
    iter
  end
end

#scan_prefix(prefix, durability_filter: nil, dirty: nil, read_ahead_bytes: nil, cache_blocks: nil, max_fetch_tasks: nil, order: nil, from: nil, to: nil) ⇒ Iterator

Scan all keys with a given prefix.

Examples:

Scan all user keys

db.scan_prefix("user:") do |key, value|
  puts "#{key}: #{value}"
end

Scan a sub-range within a prefix

# keys "user:100" (inclusive) up to "user:200" (exclusive)
db.scan_prefix("user:", from: "100", to: "200")

Parameters:

  • prefix (String) —

    The key prefix to scan

  • durability_filter (String, nil) (defaults to: nil) —

    Filter by durability level ("remote" or "memory")

  • dirty (Boolean, nil) (defaults to: nil) —

    Whether to include uncommitted data

  • read_ahead_bytes (Integer, nil) (defaults to: nil) —

    Number of bytes to read ahead

  • cache_blocks (Boolean, nil) (defaults to: nil) —

    Whether to cache blocks

  • max_fetch_tasks (Integer, nil) (defaults to: nil) —

    Maximum number of fetch tasks

  • order (Symbol, String, nil) (defaults to: nil) —

    Iteration order (:asc/:ascending or :desc/:descending)

  • from (String, nil) (defaults to: nil) —

    Inclusive lower bound suffix, appended to the prefix, to start scanning from (e.g. prefix "user:" with from "100" starts at "user:100"). Defaults to the start of the prefix.

  • to (String, nil) (defaults to: nil) —

    Exclusive upper bound suffix, appended to the prefix, to stop scanning at. Defaults to the end of the prefix.

Returns:

  • (Iterator) —

    An iterator over key-value pairs



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
 
# File 'lib/slatedb/database.rb', line 265

def scan_prefix(prefix, durability_filter: nil, dirty: nil,
                read_ahead_bytes: nil, cache_blocks: nil, max_fetch_tasks: nil, order: nil,
                from: nil, to: nil, &)
  opts = scan_options(
    durability_filter: durability_filter,
    dirty: dirty,
    read_ahead_bytes: read_ahead_bytes,
    cache_blocks: cache_blocks,
    max_fetch_tasks: max_fetch_tasks,
    order: order
  )
  opts[:subrange_from] = from if from
  opts[:subrange_to] = to if to

  iter = if opts.empty?
           _scan_prefix(prefix)
         else
           _scan_prefix_with_options(prefix, opts)
         end

  if block_given?
    iter.each(&)
  else
    iter
  end
end

#snapshot {|snapshot| ... } ⇒ Snapshot, Object

Create a snapshot for consistent reads.

Examples:

Manual snapshot management

snapshot = db.snapshot
snapshot.get("key")
snapshot.close

Block-based snapshot (auto-close)

db.snapshot do |snap|
  snap.get("key1")
  snap.get("key2")
end # automatically closed

Yields:

  • (snapshot) —

    If a block is given, yields the snapshot and auto-closes

Returns:

  • (Snapshot, Object) —

    The snapshot (or block result if block given)



465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
 
# File 'lib/slatedb/database.rb', line 465

def snapshot
  snap = _snapshot

  if block_given?
    begin
      yield snap
    ensure
      begin
        snap.close
      rescue StandardError
        nil
      end
    end
  else
    snap
  end
end

#transaction(isolation: nil) {|txn| ... } ⇒ Object

Execute a block within a transaction.

The transaction is automatically committed if the block succeeds, or rolled back if an exception is raised.

Examples:

result = db.transaction do |txn|
  old_val = txn.get("counter") || "0"
  new_val = (old_val.to_i + 1).to_s
  txn.put("counter", new_val)
  new_val
end

Parameters:

  • isolation (Symbol, String) (defaults to: nil) —

    Isolation level (:snapshot or :serializable)

Yields:

  • (txn) —

    Yields the transaction to the block

Returns:

  • (Object) —

    The result of the block



433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
 
# File 'lib/slatedb/database.rb', line 433

def transaction(isolation: nil)
  txn = begin_transaction(isolation: isolation)
  begin
    result = yield txn
    txn.commit
    result
  rescue StandardError
    begin
      txn.rollback
    rescue StandardError
      nil
    end
    raise
  end
end

#write(batch, await_durable: nil, seqnum: nil) ⇒ void

This method returns an undefined value.

Write a batch of operations atomically.

Examples:

Write a batch

batch = SlateDb::WriteBatch.new
batch.put("key1", "value1")
batch.put("key2", "value2")
batch.delete("key3")
db.write(batch)

Using batch block helper

db.batch do |b|
  b.put("key1", "value1")
  b.put("key2", "value2")
end

Parameters:

  • batch (WriteBatch) —

    The batch to write

  • await_durable (Boolean) (defaults to: nil) —

    Whether to wait for durability (default: true)

  • seqnum (Integer, nil) (defaults to: nil) —

    User-supplied sequence number applied to the batch. See #put for semantics. (Requires SlateDB >= 0.13.0)



327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/slatedb/database.rb', line 327

def write(batch, await_durable: nil, seqnum: nil)
  opts = {}
  opts[:await_durable] = await_durable unless await_durable.nil?
  opts[:seqnum] = seqnum if seqnum

  if opts.empty?
    _write(batch)
  else
    _write_with_options(batch, opts)
  end
end