Asynchronous Indexing Pseudo Code

[chubei]

Asynchronous Indexing Pseudo Code

This is mainly for clean my thoughts before starting implementation. But it's welcomed if anyone is interested and want to review the algorithm. Reviewing is easier in pseudo code.

Main Environment

struct RecordMetadata {
    /// The record id. Consistent across `insert`s and `delete`s.
    record_id: u64,
    /// The operation id of the latest `Insert` operation. `None` if the record is deleted.
    insert_operation_id: Option<u64>,
}

enum Operation {
    Delete {
        /// The operation id of an `Insert` operation, which must exist.
        operation_id: u64,
    },
    Insert {
        record: RecordWithId,
    },
}

struct MainEnvironment {
    /// Record primary key -> RecordMetadata, empty if schema has no primary key.
    /// Length always increases.
    primary_key_to_metadata: HashMap<Vec<u8>, RecordMetadata>,
    /// The next operation id. Monotonically increasing.
    next_operation_id: u64,
    /// Operation_id -> operation.
    operation_id_to_operation: HashMap<u64, Operation>,
    /// Operation ids of `Delete`s, append only. Used to filter out deleted records in query.
    deleted_operation_ids: HashSet<u64>,
}

Secondary Index Environment

struct SecondaryIndexEnvironment {
    /// The next operation id to be indexed. Monotonically increasing.
    next_operation_id: u64,
    /// Index key -> operation ids, operation ids must be the ids of `Insert`s.
    secondary_key_to_operation_id: HashMap<Vec<u8>, Vec<u64>>,
}

Insert to Main Environment

fn insert(env: &mut MainEnvironment, record: &Record) {
    // Generate operation id.
    let operation_id = env.next_operation_id;
    env.next_operation_id += 1;
    // Calculate record id.
    let record_id = if let Some(primary_key) = record.primary_key {
        // Get or generate record id from `primary_key_to_metadata`.
        let record_id = match env.primary_key_to_metadata.get(primary_key) {
            // Primary key is never inserted before. Generate new id from `primary_key_to_metadata`.
            None => env.primary_key_to_metadata.len(),
            // This primary key is present. It's an error.
            Some(RecordMetadata { insert_operation_id: Some(_), .. }) => return Err(PrimaryKeyExists),
            // This primary key was deleted. Use the record id from its first insertion.
            Some(RecordMetadata { record_id, .. }) => record_id,
        };
        // Update `primary_key_to_metadata`.
        env.primary_key_to_metadata.insert_overwrite(
            primary_key,
            RecordMetadata { record_id, insert_operation_id: Some(operation_id) }
        )?;
        record_id
    } else {
        // If the record has no primary key, record id is operation id.
        operation_id
    };
    // Record operation. The operation id must not exist.
    env.operation_id_to_operation.insert(operation_id, Operation::Insert { record: RecordWithId::new(record, record_id) }).unwrap();
}

Delete from Main Environment

fn delete(env: &mut MainEnvironment, primary_key: &[u8]) {
    /// Find operation id by primary key.
    let Some(
        RecordMetadata { record_id, insert_operation_id: Some(insert_operation_id) }
    ) = env.primary_key_to_metadata.get(primary_key) else {
        return Err(PrimaryKeyNotFound),
    };
    // Record deleted operation id.
    env.deleted_operation_ids.insert(insert_operation_id);
    // Generate operation id.
    let operation_id = env.next_operation_id;
    env.next_operation_id += 1;
    // Record operation. The operation id must not exist.
    env.operation_id_to_operation.insert(operation_id, Operation::Delete { operation_id: insert_operation_id }).unwrap();
    // Update `primary_key_to_metadata`.
    env.primary_key_to_metadata.insert_overwrite(primary_key, RecordMetadata { record_id, insert_operation_id: None });
}

Index

fn index(index_env: &mut SecondaryIndexEnvironment, main_env: &MainEnvironment) {
    // Start from `index_env.next_operation_id`.
    while index_env.next_operation_id < main_env.next_operation_id {
        // Get operation by operation id. The operation must exist.
        let operation_id = index_env.next_operation_id;
        let Some(operation) = main_env.operation_id_to_operation.get(operation_id) else { panic!() };
        match operation {
            Operation::Delete { operation_id } => {
                // If the operation is a `Delete`, find the deleted record, which must exist.
                let Some(Operation::Insert { record }) = main_env.operation_id_to_operation.get(operation_id) else { panic!() };
                // Calculate the secondary key.
                let key = calculate_secondary_key(record.record);
                // Remove operation id by secondary key. Secondary key and operation id pair must exist.
                index_env.secondary_key_to_operation_id.get_mut(key).unwrap().remove(operation_id).unwrap();
            }
            Operation::Insert { record } => {
                // If the operation is an `Insert`, calculate the secondary key.
                let key = calculate_secondary_key(record.record);
                // Insert operation id to `secondary_key_to_operation_id`.
                index_env.secondary_key_to_operation_id.entry(key).or_default().push(operation_id);
            }
        }
        index_env.next_operation_id += 1;
    }
}

Query

fn query(
    index_env: &SecondaryIndexEnvironment,
    main_env: &MainEnvironment,
    start_key: &[u8],
    end_key: &[u8],
) -> impl Iterator<Item = RecordWithId> {
    index_env.secondary_key_to_operation_id
        .range(start_key, end_key)
        .flatten()
        .filter_map(|operation_id| {
            if main_env.deleted_operation_ids.contains(operation_id) {
                None
            } else {
                let Some(Operation::Insert { record }) = main_env.operation_id_to_operation.get(operation_id) else { panic!() };
                Some(record)
            }
        })
}

Operation Log Cleanup

/// `max_operation_id` must be less than all secondary index's `next_operation_id`.
fn cleanup(operation_id_to_operation: &mut HashMap<u64, Operation>, max_operation_id: u64) {
    // Traverse all operation ids in reverse order.
    for operation_id in (0..=max_operation_id).into_iter().rev() {
        if let Entry::Occupied(entry) = operation_id_to_operation.entry(operation_id) {
            if let Operation::Delete { operation_id } = entry.get() {
                // If the operation is a `Delete`, remove this operation and the deleted operation.
                entry.remove();
                // The deleted operation must exist.
                operation_id_to_operation.remove(operation_id).unwrap();
            }
        }
    }
}