feoxdb/core/store/
internal.rs

1use ahash::RandomState;
2use bytes::Bytes;
3use scc::HashMap;
4use std::sync::atomic::Ordering;
5use std::sync::Arc;
6
7use crate::constants::Operation;
8use crate::core::record::Record;
9use crate::error::{FeoxError, Result};
10use crate::storage::write_buffer::WriteBuffer;
11
12use super::FeoxStore;
13
14impl FeoxStore {
15    pub(super) fn update_record_with_ttl(
16        &self,
17        old_record: &Record,
18        value: &[u8],
19        timestamp: u64,
20        ttl_expiry: u64,
21    ) -> Result<bool> {
22        let new_record = if ttl_expiry > 0 && self.enable_ttl {
23            Arc::new(Record::new_with_timestamp_ttl(
24                old_record.key.clone(),
25                value.to_vec(),
26                timestamp,
27                ttl_expiry,
28            ))
29        } else {
30            Arc::new(Record::new(
31                old_record.key.clone(),
32                value.to_vec(),
33                timestamp,
34            ))
35        };
36
37        let old_value_len = old_record.value_len;
38        let old_size = old_record.calculate_size();
39        let new_size = self.calculate_record_size(old_record.key.len(), value.len());
40
41        let old_record_arc =
42            if let Some(entry) = self.hash_table.read(&old_record.key, |_, v| v.clone()) {
43                entry
44            } else {
45                return Err(FeoxError::KeyNotFound);
46            };
47
48        let key_vec = new_record.key.clone();
49
50        old_record_arc.refcount.store(0, Ordering::Release);
51        self.hash_table
52            .upsert(key_vec.clone(), Arc::clone(&new_record));
53        self.tree.insert(key_vec.clone(), Arc::clone(&new_record));
54
55        if new_size > old_size {
56            self.stats
57                .memory_usage
58                .fetch_add(new_size - old_size, Ordering::AcqRel);
59        } else {
60            self.stats
61                .memory_usage
62                .fetch_sub(old_size - new_size, Ordering::AcqRel);
63        }
64
65        // Only do cache and persistence operations if not in memory-only mode
66        if !self.memory_only {
67            if self.enable_caching {
68                if let Some(ref cache) = self.cache {
69                    cache.remove(&key_vec);
70                }
71            }
72
73            if let Some(ref wb) = self.write_buffer {
74                wb.add_write(Operation::Update, Arc::clone(&new_record), old_value_len)?;
75                wb.add_write(Operation::Delete, old_record_arc, old_value_len)?;
76            }
77        }
78
79        Ok(false)
80    }
81
82    /// Update an existing record with a new value (Bytes version for zero-copy).
83    pub(super) fn update_record_with_ttl_bytes(
84        &self,
85        old_record: &Record,
86        value: Bytes,
87        timestamp: u64,
88        ttl_expiry: u64,
89    ) -> Result<bool> {
90        let new_record = if ttl_expiry > 0 && self.enable_ttl {
91            Arc::new(Record::new_from_bytes_with_ttl(
92                old_record.key.clone(),
93                value,
94                timestamp,
95                ttl_expiry,
96            ))
97        } else {
98            Arc::new(Record::new_from_bytes(
99                old_record.key.clone(),
100                value,
101                timestamp,
102            ))
103        };
104
105        let old_value_len = old_record.value_len;
106        let old_size = old_record.calculate_size();
107        let new_size = new_record.calculate_size();
108
109        let old_record_arc =
110            if let Some(entry) = self.hash_table.read(&old_record.key, |_, v| v.clone()) {
111                entry
112            } else {
113                return Err(FeoxError::KeyNotFound);
114            };
115
116        let key_vec = new_record.key.clone();
117
118        old_record_arc.refcount.store(0, Ordering::Release);
119        self.hash_table
120            .upsert(key_vec.clone(), Arc::clone(&new_record));
121        self.tree.insert(key_vec.clone(), Arc::clone(&new_record));
122
123        if new_size > old_size {
124            self.stats
125                .memory_usage
126                .fetch_add(new_size - old_size, Ordering::AcqRel);
127        } else {
128            self.stats
129                .memory_usage
130                .fetch_sub(old_size - new_size, Ordering::AcqRel);
131        }
132
133        // Only do cache and persistence operations if not in memory-only mode
134        if !self.memory_only {
135            if self.enable_caching {
136                if let Some(ref cache) = self.cache {
137                    cache.remove(&key_vec);
138                }
139            }
140
141            if let Some(ref wb) = self.write_buffer {
142                wb.add_write(Operation::Update, Arc::clone(&new_record), old_value_len)?;
143                wb.add_write(Operation::Delete, old_record_arc, old_value_len)?;
144            }
145        }
146
147        Ok(false)
148    }
149
150    /// Get access to hash table (for TTL cleaner)
151    pub(crate) fn get_hash_table(&self) -> &HashMap<Vec<u8>, Arc<Record>, RandomState> {
152        &self.hash_table
153    }
154
155    /// Remove from tree (for TTL cleaner)
156    pub(crate) fn remove_from_tree(&self, key: &[u8]) {
157        self.tree.remove(key);
158    }
159
160    /// Get write buffer (for TTL cleaner)
161    pub(crate) fn get_write_buffer(&self) -> Option<&Arc<WriteBuffer>> {
162        self.write_buffer.as_ref()
163    }
164}
feoxdb/core/store/internal.rs

feoxdb/core/store/
internal.rs
