Design Recipes¶

import fdb
fdb.api_version(730)

user = fdb.Subspace(('user',))
index = fdb.Subspace(('zipcode_index',))

@fdb.transactional
def set_user(tr, ID, name, zipcode):
    """Store user and update zipcode index atomically."""
    tr[user[ID][zipcode]] = name
    tr[index[zipcode][ID]] = ''

@fdb.transactional
def get_user(tr, ID):
    """Retrieve user by primary key."""
    for k, v in tr[user[ID].range()]:
        return v
    return None

@fdb.transactional
def get_user_ids_in_region(tr, zipcode):
    """Query index to find all users in a zipcode."""
    return [index.unpack(k)[1] for k, _ in tr[index[zipcode].range()]]

import fdb
fdb.api_version(730)

table = fdb.Subspace(('T',))
row_index = table['R']
col_index = table['C']

def _pack(value):
    return fdb.tuple.pack((value,))

def _unpack(value):
    return fdb.tuple.unpack(value)[0]

@fdb.transactional
def table_set_cell(tr, row, column, value):
    """Set a cell, updating both row and column indexes."""
    tr[row_index[row][column]] = _pack(value)
    tr[col_index[column][row]] = _pack(value)

@fdb.transactional
def table_get_cell(tr, row, column):
    """Get a single cell value."""
    return _unpack(tr[row_index[row][column]])

@fdb.transactional
def table_get_row(tr, row):
    """Get all cells in a row with a single range read."""
    cols = {}
    for k, v in tr[row_index[row].range()]:
        r, c = row_index.unpack(k)
        cols[c] = _unpack(v)
    return cols

@fdb.transactional
def table_get_col(tr, col):
    """Get all cells in a column with a single range read."""
    rows = {}
    for k, v in tr[col_index[col].range()]:
        c, r = col_index.unpack(k)
        rows[r] = _unpack(v)
    return rows

@fdb.transactional
def table_set_row(tr, row, cols):
    """Replace an entire row."""
    del tr[row_index[row].range()]
    for c, v in cols.items():
        table_set_cell(tr, row, c, v)

import fdb
import os

fdb.api_version(730)

queue = fdb.Subspace(('Q',))

@fdb.transactional
def enqueue(tr, value):
    """Add item to the queue (conflict-free via snapshot read)."""
    tr[queue[last_index(tr) + 1][os.urandom(20)]] = value

@fdb.transactional
def dequeue(tr):
    """Remove and return the first item, or None if empty."""
    item = first_item(tr)
    if item is None:
        return None
    del tr[item.key]
    return item.value

@fdb.transactional
def last_index(tr):
    """Find the highest index using a reverse range read."""
    r = queue.range()
    for key, _ in tr.snapshot.get_range(r.start, r.stop,
                                         limit=1, reverse=True):
        return queue.unpack(key)[0]
    return 0

@fdb.transactional
def first_item(tr):
    """Get the first item in the queue."""
    r = queue.range()
    for kv in tr.get_range(r.start, r.stop, limit=1):
        return kv
    return None

import fdb
import os

fdb.api_version(730)

pq = fdb.Subspace(('P',))

@fdb.transactional
def push(tr, value, priority):
    """Add item with given priority."""
    tr[pq[priority][_next_count(tr, priority)][os.urandom(20)]] = value

@fdb.transactional
def _next_count(tr, priority):
    """Get next count for a priority level (snapshot read for conflict-free)."""
    r = pq[priority].range()
    for key, _ in tr.snapshot.get_range(r.start, r.stop,
                                         limit=1, reverse=True):
        return pq[priority].unpack(key)[0] + 1
    return 0

@fdb.transactional
def pop(tr, max=False):
    """Remove and return min (default) or max priority item."""
    r = pq.range()
    for item in tr.get_range(r.start, r.stop, limit=1, reverse=max):
        del tr[item.key]
        return item.value
    return None

@fdb.transactional
def peek(tr, max=False):
    """Return min (default) or max priority item without removing."""
    r = pq.range()
    for item in tr.get_range(r.start, r.stop, limit=1, reverse=max):
        return item.value
    return None

# Convenience functions
def pop_min(tr): return pop(tr, max=False)
def pop_max(tr): return pop(tr, max=True)
def peek_min(tr): return peek(tr, max=False)
def peek_max(tr): return peek(tr, max=True)

import fdb
fdb.api_version(730)

vector = fdb.Subspace(('V',))

@fdb.transactional
def vector_get(tr, index):
    """Get element at index."""
    return tr[vector[index]]

@fdb.transactional
def vector_set(tr, index, value):
    """Set element at index."""
    tr[vector[index]] = value

@fdb.transactional
def vector_push(tr, value):
    """Append to end of vector."""
    tr[vector[vector_size(tr)]] = value

@fdb.transactional
def vector_size(tr):
    """Get vector length."""
    r = vector.range()
    for key, _ in tr.get_range(r.start, r.stop, limit=1, reverse=True):
        return vector.unpack(key)[0] + 1
    return 0

@fdb.transactional
def vector_scan(tr, start=0, end=None):
    """Iterate over vector elements."""
    if end is None:
        r = vector.range((start,))
    else:
        r = vector.range((start,), (end,))
    return [v for k, v in tr[r]]

@fdb.transactional
def vector_truncate(tr, new_size):
    """Remove elements from new_size onwards."""
    r = vector.range()
    del tr[vector[new_size].key():r.stop]

import fdb
import json
import random
import itertools

fdb.api_version(730)

doc_space = fdb.Subspace(('D',))

EMPTY_OBJECT = -2
EMPTY_ARRAY = -1

def to_tuples(item):
    """Convert nested structure to list of path tuples."""
    if item == {}:
        return [(EMPTY_OBJECT, None)]
    elif item == []:
        return [(EMPTY_ARRAY, None)]
    elif isinstance(item, dict):
        return [(k,) + sub for k, v in item.items() for sub in to_tuples(v)]
    elif isinstance(item, list):
        return [(i,) + sub for i, v in enumerate(item) for sub in to_tuples(v)]
    else:
        return [(item,)]

def from_tuples(tuples):
    """Reconstruct nested structure from path tuples."""
    if not tuples:
        return {}
    first = tuples[0]
    if len(first) == 1:
        return first[0]
    if first == (EMPTY_OBJECT, None):
        return {}
    if first == (EMPTY_ARRAY, None):
        return []

    groups = [list(g) for k, g in itertools.groupby(tuples, lambda t: t[0])]
    if first[0] == 0:  # array
        return [from_tuples([t[1:] for t in g]) for g in groups]
    else:  # object
        return {g[0][0]: from_tuples([t[1:] for t in g]) for g in groups}

@fdb.transactional
def insert_doc(tr, doc):
    """Insert a document, returning its ID."""
    if isinstance(doc, str):
        doc = json.loads(doc)
    if 'doc_id' not in doc:
        doc['doc_id'] = random.randint(0, 100000000)
    for tup in to_tuples(doc):
        tr[doc_space.pack((doc['doc_id'],) + tup[:-1])] = \
            fdb.tuple.pack((tup[-1],))
    return doc['doc_id']

@fdb.transactional
def get_doc(tr, doc_id, prefix=()):
    """Get document or subdocument by path prefix."""
    v = tr[doc_space.pack((doc_id,) + prefix)]
    if v.present():
        return from_tuples([prefix + fdb.tuple.unpack(v)])
    else:
        return from_tuples([
            doc_space.unpack(k)[1:] + fdb.tuple.unpack(v)
            for k, v in tr[doc_space.range((doc_id,) + prefix)]
        ])

import fdb
fdb.api_version(730)

CHUNK_SIZE = 10000  # 10KB chunks for optimal performance

blob_subspace = fdb.Subspace(('blob',))

@fdb.transactional
def write_blob(tr, blob_id, data):
    """Write a blob by chunking it into multiple keys."""
    # Clear any existing blob data
    del tr[blob_subspace[blob_id].range()]

    # Write chunks
    for offset in range(0, len(data), CHUNK_SIZE):
        chunk = data[offset:offset + CHUNK_SIZE]
        tr[blob_subspace[blob_id][offset]] = chunk

@fdb.transactional
def read_blob(tr, blob_id):
    """Read entire blob by combining chunks."""
    chunks = []
    for k, v in tr[blob_subspace[blob_id].range()]:
        chunks.append(v)
    return b''.join(chunks)

@fdb.transactional
def read_blob_range(tr, blob_id, start, length):
    """Read a range of bytes from a blob."""
    # Find which chunks we need
    start_chunk = (start // CHUNK_SIZE) * CHUNK_SIZE
    end_byte = start + length

    result = b''
    for k, v in tr[blob_subspace[blob_id].range()]:
        offset = blob_subspace[blob_id].unpack(k)[0]
        if offset + len(v) <= start:
            continue
        if offset >= end_byte:
            break

        # Calculate slice within this chunk
        chunk_start = max(0, start - offset)
        chunk_end = min(len(v), end_byte - offset)
        result += v[chunk_start:chunk_end]

    return result

@fdb.transactional
def delete_blob(tr, blob_id):
    """Delete a blob."""
    del tr[blob_subspace[blob_id].range()]

@fdb.transactional
def blob_size(tr, blob_id):
    """Get blob size without reading all data."""
    total = 0
    for k, v in tr[blob_subspace[blob_id].range()]:
        total += len(v)
    return total

import fdb
fdb.api_version(730)

label_z = fdb.Subspace(('L',))  # label -> z-value
z_label = fdb.Subspace(('Z',))  # z-value -> label

def xy_to_z(x, y):
    """Convert 2D coordinates to Z-order value (Morton code).

    Interleaves bits of x and y to create a single value
    that preserves spatial locality.
    """
    z = 0
    for i in range(32):
        z |= ((x >> i) & 1) << (2 * i)
        z |= ((y >> i) & 1) << (2 * i + 1)
    return z

def z_to_xy(z):
    """Convert Z-order value back to 2D coordinates."""
    x = y = 0
    for i in range(32):
        x |= ((z >> (2 * i)) & 1) << i
        y |= ((z >> (2 * i + 1)) & 1) << i
    return x, y

@fdb.transactional
def set_location(tr, label, x, y):
    """Set or update an item's location."""
    # Remove old location if it exists
    old_z = None
    for k, _ in tr[label_z[label].range()]:
        old_z = label_z.unpack(k)[1]
        break

    if old_z is not None:
        del tr[label_z[label][old_z]]
        del tr[z_label[old_z][label]]

    # Set new location
    new_z = xy_to_z(x, y)
    tr[label_z[label][new_z]] = b''
    tr[z_label[new_z][label]] = b''

@fdb.transactional
def get_location(tr, label):
    """Get an item's current location."""
    for k, _ in tr[label_z[label].range()]:
        z = label_z.unpack(k)[1]
        return z_to_xy(z)
    return None

@fdb.transactional
def get_in_box(tr, x_min, y_min, x_max, y_max):
    """Find all labels within a bounding box.

    Note: Z-order ranges are an approximation. Some false positives
    outside the box may be returned and need filtering.
    """
    results = []
    z_min = xy_to_z(x_min, y_min)
    z_max = xy_to_z(x_max, y_max)

    # Scan the Z-range and filter to exact box
    for k, _ in tr[z_label[z_min] : z_label[z_max + 1]]:
        z, label = z_label.unpack(k)
        x, y = z_to_xy(z)
        if x_min <= x <= x_max and y_min <= y <= y_max:
            results.append((label, x, y))
    return results

import fdb
fdb.api_version(730)

class Workspace:
    """Blue/green workspace for atomic batch updates.

    Usage:
        workspace = Workspace(working_dir, db)

        # Read current data
        for k, v in db[workspace.current.range()]:
            process(k, v)

        # Build new version
        with workspace as new_space:
            db[new_space['key1']] = 'value1'
            db[new_space['key2']] = 'value2'
        # On exit, new_space atomically becomes current
    """

    def __init__(self, directory, db):
        self.dir = directory
        self.db = db

    @property
    def current(self):
        """Get the current (active) workspace."""
        return self.dir.create_or_open(self.db, ('current',))

    def __enter__(self):
        """Create a new workspace for staging updates."""
        return self.dir.create_or_open(self.db, ('new',))

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Atomically swap new workspace to current."""
        if exc_type is None:  # Only swap on success
            self._swap(self.db)

    @fdb.transactional
    def _swap(self, tr):
        """Atomic swap: remove current, rename new to current."""
        # Remove old current
        if self.dir.exists(tr, ('current',)):
            self.dir.remove(tr, ('current',))
        # Move new to current
        self.dir.move(tr, ('new',), ('current',))


# Example usage
def refresh_catalog(db, new_items):
    """Replace entire catalog atomically."""
    working_dir = fdb.directory.create_or_open(db, ('catalog_workspace',))
    workspace = Workspace(working_dir, db)

    with workspace as staging:
        # Clear staging area
        del db[staging.range()]

        # Write all new items
        for item_id, data in new_items.items():
            db[staging[item_id]] = fdb.tuple.pack((data,))

    # Swap complete - readers now see new catalog