You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: mod.rs
    path of file: ./repos/arroyo/arroyo-state/src/tables
    
    the code of the file until where you have to start completion: use crate::DataOperation;
use arroyo_rpc::grpc::TableType;
use std::collections::{BTreeMap, HashMap};
use std::time::SystemTime;

pub mod global_keyed_map;
pub mod key_time_multi_map;
pub mod keyed_map;
pub mod time_key_map;

pub enum Compactor {
    TimeKeyMap,
    KeyTimeMultiMap,
}

pub struct DataTuple<K, V> {
    pub timestamp: SystemTime,
    pub key: K,
    pub value: Option<V>,
    pub operation: DataOperation,
}

/// BlindDataTuple's key and value are not decoded
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct BlindDataTuple {
    pub key_hash: u64,
    pub timestamp: SystemTime,
    pub key: Vec<u8>,
    pub value: Vec<u8>,
    pub operation: DataOperation,
}

impl Compactor {
    pub(crate) fn for_table_type(table_type: TableType) -> Self {
        match table_type {
            TableType::Global | TableType::TimeKeyMap => Compactor::TimeKeyMap,
            TableType::KeyTimeMultiMap => Compactor::KeyTimeMultiMap,
        }
    }

    pub(crate) fn compact_tuples(&self, tuples: Vec<BlindDataTuple>) -> Vec<BlindDataTuple> {
        match self {
            Compactor::TimeKeyMap => {
                // keep only the latest entry for each key
                let mut reduced = BTreeMap::new();
                for tuple in tuples.into_iter() {
                    match tuple.operation {
                        DataOperation::Insert | DataOperation::DeleteTimeKey(_) => {}
    
    
    async fn test_key_time_multi_map_compaction() {
        let t1 = SystemTime::now();
        let t2 = t1 + Duration::from_secs(1);
        let t3 = t2 + Duration::from_secs(1);

        let k1 = "k1".as_bytes().to_vec();
        let v1 = "v1".as_bytes().to_vec();

        let insert_1 = BlindDataTuple {
            key_hash: 123,
            timestamp: t1,
            key: k1.clone(),
            value: v1.clone(),
            operation: DataOperation::Insert,
        };

        let insert_2 = BlindDataTuple {
            key_hash: 123,
            timestamp: t2,
            key: k1.clone(),
            value: v1.clone(),
            operation: DataOperation::Insert,
        };

        let delete_all = BlindDataTuple {
            key_hash: 123,
            timestamp: t2,
            key: k1.clone(),
            value: v1.clone(),
            operation: DataOperation::DeleteTimeRange(DeleteTimeRangeOperation {
                key: k1.clone(),
                start: t1,
                end: t3,
            }),
        };

        let insert_3 = BlindDataTuple {
            key_hash: 123,
            timestamp: t1,
            key: k1.clone(),
            value: v1.clone(),
            operation: DataOperation::Insert,
        };

        let tuples_in = vec![
            insert_1.clone(),
            insert_2.clone(),
            delete_all.clone(),
            insert_3.clone(),
        ];

        let tuples_out = Compactor::KeyTimeMultiMap.compact_tuples(tuples_in);
        assert_eq!(vec![delete_all, insert_3], tuples_out);

        // test idempotence
        assert_eq!(
            tuples_out.clone(),
            Compactor::KeyTimeMultiMap.compact_tuples(tuples_out),
        );
    }