feoxdb/core/store/operations.rs
1use bytes::Bytes;
2use std::sync::atomic::Ordering;
3use std::sync::Arc;
4use std::time::{SystemTime, UNIX_EPOCH};
5
6use crate::constants::*;
7use crate::core::record::Record;
8use crate::error::{FeoxError, Result};
9
10use super::FeoxStore;
11
12impl FeoxStore {
13 /// Insert or update a key-value pair.
14 ///
15 /// If the key already exists with a TTL, the TTL is removed (key becomes permanent).
16 /// To preserve or set TTL, use `insert_with_ttl()` instead.
17 ///
18 /// # Arguments
19 ///
20 /// * `key` - The key to insert
21 /// * `value` - The value to store
22 /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
23 ///
24 /// # Returns
25 ///
26 /// Returns `Ok(true)` if a new key was inserted, `Ok(false)` if an existing key was updated.
27 ///
28 /// # Errors
29 ///
30 /// * `InvalidKey` - Key is empty or too large
31 /// * `InvalidValue` - Value is too large
32 /// * `OlderTimestamp` - Timestamp is not newer than existing record
33 /// * `OutOfMemory` - Memory limit exceeded
34 ///
35 /// # Example
36 ///
37 /// ```rust
38 /// # use feoxdb::FeoxStore;
39 /// # fn main() -> feoxdb::Result<()> {
40 /// # let store = FeoxStore::new(None)?;
41 /// store.insert(b"user:123", b"{\"name\":\"Mehran\"}")?;
42 /// # Ok(())
43 /// # }
44 /// ```
45 ///
46 /// # Performance
47 ///
48 /// * Memory mode: ~600ns
49 /// * Persistent mode: ~800ns (buffered write)
50 pub fn insert(&self, key: &[u8], value: &[u8]) -> Result<bool> {
51 self.insert_with_timestamp(key, value, None)
52 }
53
54 /// Insert or update a key-value pair with explicit timestamp.
55 ///
56 /// This is the advanced version that allows manual timestamp control for
57 /// conflict resolution. Most users should use `insert()` instead.
58 ///
59 /// # Arguments
60 ///
61 /// * `key` - The key to insert
62 /// * `value` - The value to store
63 /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
64 ///
65 /// # Errors
66 ///
67 /// * `OlderTimestamp` - Timestamp is not newer than existing record
68 pub fn insert_with_timestamp(
69 &self,
70 key: &[u8],
71 value: &[u8],
72 timestamp: Option<u64>,
73 ) -> Result<bool> {
74 self.insert_with_timestamp_and_ttl_internal(key, value, timestamp, 0)
75 }
76
77 /// Insert or update a key-value pair using zero-copy Bytes.
78 ///
79 /// This method avoids copying the value data by directly using the Bytes type,
80 /// which provides reference-counted zero-copy semantics. Useful when inserting
81 /// data that was already read from network or disk as Bytes.
82 ///
83 /// If the key already exists with a TTL, the TTL is removed (key becomes permanent).
84 /// To preserve or set TTL, use `insert_bytes_with_ttl()` instead.
85 ///
86 /// # Arguments
87 ///
88 /// * `key` - The key to insert
89 /// * `value` - The value to store as Bytes
90 ///
91 /// # Returns
92 ///
93 /// Returns `Ok(true)` if a new key was inserted, `Ok(false)` if an existing key was updated.
94 ///
95 /// # Errors
96 ///
97 /// * `InvalidKey` - Key is empty or too large
98 /// * `InvalidValue` - Value is too large
99 /// * `OlderTimestamp` - Timestamp is not newer than existing record
100 /// * `OutOfMemory` - Memory limit exceeded
101 ///
102 /// # Example
103 ///
104 /// ```rust
105 /// # use feoxdb::FeoxStore;
106 /// # use bytes::Bytes;
107 /// # fn main() -> feoxdb::Result<()> {
108 /// # let store = FeoxStore::new(None)?;
109 /// let data = Bytes::from_static(b"{\"name\":\"Mehran\"}");
110 /// store.insert_bytes(b"user:123", data)?;
111 /// # Ok(())
112 /// # }
113 /// ```
114 ///
115 /// # Performance
116 ///
117 /// * Memory mode: ~600ns (avoids value copy)
118 /// * Persistent mode: ~800ns (buffered write, avoids value copy)
119 pub fn insert_bytes(&self, key: &[u8], value: Bytes) -> Result<bool> {
120 self.insert_bytes_with_timestamp(key, value, None)
121 }
122
123 /// Insert or update a key-value pair using zero-copy Bytes with explicit timestamp.
124 ///
125 /// This is the advanced version that allows manual timestamp control for
126 /// conflict resolution. Most users should use `insert_bytes()` instead.
127 ///
128 /// # Arguments
129 ///
130 /// * `key` - The key to insert
131 /// * `value` - The value to store as Bytes
132 /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
133 ///
134 /// # Errors
135 ///
136 /// * `OlderTimestamp` - Timestamp is not newer than existing record
137 pub fn insert_bytes_with_timestamp(
138 &self,
139 key: &[u8],
140 value: Bytes,
141 timestamp: Option<u64>,
142 ) -> Result<bool> {
143 self.insert_bytes_with_timestamp_and_ttl_internal(key, value, timestamp, 0)
144 }
145
146 pub(super) fn insert_with_timestamp_and_ttl_internal(
147 &self,
148 key: &[u8],
149 value: &[u8],
150 timestamp: Option<u64>,
151 ttl_expiry: u64,
152 ) -> Result<bool> {
153 let start = std::time::Instant::now();
154 let timestamp = match timestamp {
155 Some(0) | None => self.get_timestamp(),
156 Some(ts) => ts,
157 };
158 self.validate_key_value(key, value)?;
159
160 // Check for existing record
161 let is_update = self.hash_table.contains(key);
162 let existing_record = self.hash_table.read(key, |_, v| v.clone());
163 if let Some(existing_record) = existing_record {
164 let existing_ts = existing_record.timestamp;
165 let existing_clone = existing_record;
166
167 if timestamp < existing_ts {
168 return Err(FeoxError::OlderTimestamp);
169 }
170
171 // Update existing record
172 return self.update_record_with_ttl(&existing_clone, value, timestamp, ttl_expiry);
173 }
174
175 let record_size = self.calculate_record_size(key.len(), value.len());
176 if !self.check_memory_limit(record_size) {
177 return Err(FeoxError::OutOfMemory);
178 }
179
180 // Create new record with TTL if specified and TTL is enabled
181 let record = if ttl_expiry > 0 && self.enable_ttl {
182 self.stats.keys_with_ttl.fetch_add(1, Ordering::Relaxed);
183 Arc::new(Record::new_with_timestamp_ttl(
184 key.to_vec(),
185 value.to_vec(),
186 timestamp,
187 ttl_expiry,
188 ))
189 } else {
190 Arc::new(Record::new(key.to_vec(), value.to_vec(), timestamp))
191 };
192
193 let key_vec = record.key.clone();
194
195 // Insert into hash table
196 self.hash_table.upsert(key_vec.clone(), Arc::clone(&record));
197
198 // Insert into lock-free skip list for ordered access
199 self.tree.insert(key_vec, Arc::clone(&record));
200
201 // Update statistics
202 self.stats.record_count.fetch_add(1, Ordering::AcqRel);
203 self.stats
204 .memory_usage
205 .fetch_add(record_size, Ordering::AcqRel);
206 self.stats
207 .record_insert(start.elapsed().as_nanos() as u64, is_update);
208
209 // Only do persistence if not in memory-only mode
210 if !self.memory_only {
211 // Queue for persistence if write buffer exists
212 if let Some(ref wb) = self.write_buffer {
213 wb.add_write(Operation::Insert, record, 0)?;
214 }
215 }
216
217 Ok(!is_update)
218 }
219
220 /// Internal method to insert a Bytes value with timestamp and TTL (zero-copy)
221 pub(super) fn insert_bytes_with_timestamp_and_ttl_internal(
222 &self,
223 key: &[u8],
224 value: Bytes,
225 timestamp: Option<u64>,
226 ttl_seconds: u64,
227 ) -> Result<bool> {
228 let start = std::time::Instant::now();
229 // Get timestamp before any operations
230 let timestamp = match timestamp {
231 Some(0) | None => self.get_timestamp(),
232 Some(ts) => ts,
233 };
234
235 self.validate_key(key)?;
236 let value_len = value.len();
237 if value_len == 0 || value_len > MAX_VALUE_SIZE {
238 return Err(FeoxError::InvalidValueSize);
239 }
240
241 // Check for existing record
242 let is_update = self.hash_table.contains(key);
243 let existing_record = self.hash_table.read(key, |_, v| v.clone());
244 if let Some(existing_record) = existing_record {
245 let existing_ts = existing_record.timestamp;
246
247 if timestamp < existing_ts {
248 return Err(FeoxError::OlderTimestamp);
249 }
250
251 // Calculate TTL expiry
252 let ttl_expiry = if ttl_seconds > 0 && self.enable_ttl {
253 timestamp + (ttl_seconds * 1_000_000_000)
254 } else {
255 0
256 };
257
258 // Update existing record using the Bytes version
259 return self.update_record_with_ttl_bytes(
260 &existing_record,
261 value,
262 timestamp,
263 ttl_expiry,
264 );
265 }
266
267 // This point is only reached for new inserts (not updates)
268 let new_size = self.calculate_record_size(key.len(), value_len);
269 if !self.check_memory_limit(new_size) {
270 return Err(FeoxError::OutOfMemory);
271 }
272
273 // Create new record with Bytes value
274 let record = if ttl_seconds > 0 && self.enable_ttl {
275 let ttl_expiry = timestamp + (ttl_seconds * 1_000_000_000);
276 self.stats.keys_with_ttl.fetch_add(1, Ordering::Relaxed);
277 Arc::new(Record::new_from_bytes_with_ttl(
278 key.to_vec(),
279 value,
280 timestamp,
281 ttl_expiry,
282 ))
283 } else {
284 Arc::new(Record::new_from_bytes(key.to_vec(), value, timestamp))
285 };
286
287 let key_vec = record.key.clone();
288
289 // Insert into hash table
290 self.hash_table.upsert(key_vec.clone(), Arc::clone(&record));
291
292 // Insert into skip list for ordered access
293 self.tree.insert(key_vec, Arc::clone(&record));
294
295 // Update statistics
296 self.stats.record_count.fetch_add(1, Ordering::AcqRel);
297 self.stats
298 .memory_usage
299 .fetch_add(new_size, Ordering::AcqRel);
300 self.stats
301 .record_insert(start.elapsed().as_nanos() as u64, is_update);
302
303 // Only do persistence if not in memory-only mode
304 if !self.memory_only {
305 // Queue for persistence if write buffer exists
306 if let Some(ref wb) = self.write_buffer {
307 wb.add_write(Operation::Insert, record, 0)?;
308 }
309 }
310
311 Ok(!is_update)
312 }
313
314 /// Retrieve a value by key.
315 ///
316 /// # Arguments
317 ///
318 /// * `key` - The key to look up
319 /// * `expected_size` - Optional expected value size for validation
320 ///
321 /// # Returns
322 ///
323 /// Returns the value as a `Vec<u8>` if found.
324 ///
325 /// # Errors
326 ///
327 /// * `KeyNotFound` - Key does not exist
328 /// * `InvalidKey` - Key is invalid
329 /// * `SizeMismatch` - Value size doesn't match expected size
330 /// * `IoError` - Failed to read from disk (persistent mode)
331 ///
332 /// # Example
333 ///
334 /// ```rust
335 /// # use feoxdb::FeoxStore;
336 /// # fn main() -> feoxdb::Result<()> {
337 /// # let store = FeoxStore::new(None)?;
338 /// # store.insert(b"key", b"value")?;
339 /// let value = store.get(b"key")?;
340 /// assert_eq!(value, b"value");
341 /// # Ok(())
342 /// # }
343 /// ```
344 ///
345 /// # Performance
346 ///
347 /// * Memory mode: ~100ns
348 /// * Persistent mode (cached): ~150ns
349 /// * Persistent mode (disk read): ~500ns
350 pub fn get(&self, key: &[u8]) -> Result<Vec<u8>> {
351 let start = std::time::Instant::now();
352 self.validate_key(key)?;
353
354 let record = self
355 .hash_table
356 .read(key, |_, v| v.clone())
357 .ok_or(FeoxError::KeyNotFound)?;
358
359 // Check TTL expiry if TTL is enabled
360 if self.enable_ttl {
361 let ttl_expiry = record.ttl_expiry.load(Ordering::Relaxed);
362 if ttl_expiry > 0 {
363 let now = SystemTime::now()
364 .duration_since(UNIX_EPOCH)
365 .unwrap_or_default()
366 .as_nanos() as u64;
367 if now > ttl_expiry {
368 self.stats.ttl_expired_lazy.fetch_add(1, Ordering::Relaxed);
369 return Err(FeoxError::KeyNotFound);
370 }
371 }
372 }
373
374 let (value, cache_hit) = if let Some(value) = record.get_value() {
375 (value, true)
376 } else if let Some(value) = self
377 .cache
378 .as_ref()
379 .and_then(|cache| cache.get_for_record(key, &record))
380 {
381 (value, true)
382 } else {
383 (Bytes::from(self.load_value_from_disk(&record)?), false)
384 };
385
386 if !cache_hit {
387 if let Some(ref cache) = self.cache {
388 cache.insert_for_record(key.to_vec(), value.clone(), Arc::clone(&record));
389 }
390 }
391
392 self.stats
393 .record_get(start.elapsed().as_nanos() as u64, cache_hit);
394 Ok(value.to_vec())
395 }
396
397 /// Get a value by key without copying (zero-copy).
398 ///
399 /// Returns `Bytes` which avoids the memory copy that `get()` performs
400 /// when converting to `Vec<u8>`.
401 ///
402 /// # Arguments
403 ///
404 /// * `key` - The key to look up
405 ///
406 /// # Returns
407 ///
408 /// Returns the value as `Bytes` if found.
409 ///
410 /// # Example
411 ///
412 /// ```rust
413 /// # use feoxdb::FeoxStore;
414 /// # fn main() -> feoxdb::Result<()> {
415 /// # let store = FeoxStore::new(None)?;
416 /// # store.insert(b"key", b"value")?;
417 /// let bytes = store.get_bytes(b"key")?;
418 /// // Use bytes directly without copying
419 /// assert_eq!(&bytes[..], b"value");
420 /// # Ok(())
421 /// # }
422 /// ```
423 ///
424 /// # Performance
425 ///
426 /// Significantly faster than `get()` for large values:
427 /// * 100 bytes: ~15% faster
428 /// * 1KB: ~50% faster
429 /// * 10KB: ~90% faster
430 /// * 100KB: ~95% faster
431 pub fn get_bytes(&self, key: &[u8]) -> Result<Bytes> {
432 let start = std::time::Instant::now();
433 self.validate_key(key)?;
434
435 let record = self
436 .hash_table
437 .read(key, |_, v| v.clone())
438 .ok_or(FeoxError::KeyNotFound)?;
439
440 // Check TTL expiry if TTL is enabled
441 if self.enable_ttl {
442 let ttl_expiry = record.ttl_expiry.load(Ordering::Relaxed);
443 if ttl_expiry > 0 {
444 let now = SystemTime::now()
445 .duration_since(UNIX_EPOCH)
446 .unwrap_or_default()
447 .as_nanos() as u64;
448 if now > ttl_expiry {
449 self.stats.ttl_expired_lazy.fetch_add(1, Ordering::Relaxed);
450 return Err(FeoxError::KeyNotFound);
451 }
452 }
453 }
454
455 let (value, cache_hit) = if let Some(val) = record.get_value() {
456 (val, true)
457 } else if let Some(value) = self
458 .cache
459 .as_ref()
460 .and_then(|cache| cache.get_for_record(key, &record))
461 {
462 (value, true)
463 } else {
464 (Bytes::from(self.load_value_from_disk(&record)?), false)
465 };
466
467 if !cache_hit {
468 if let Some(ref cache) = self.cache {
469 cache.insert_for_record(key.to_vec(), value.clone(), Arc::clone(&record));
470 }
471 }
472
473 self.stats
474 .record_get(start.elapsed().as_nanos() as u64, cache_hit);
475 Ok(value)
476 }
477
478 /// Delete a key-value pair.
479 ///
480 /// # Arguments
481 ///
482 /// * `key` - The key to delete
483 /// * `timestamp` - Optional timestamp for conflict resolution
484 ///
485 /// # Returns
486 ///
487 /// Returns `Ok(())` if the key was deleted.
488 ///
489 /// # Errors
490 ///
491 /// * `KeyNotFound` - Key does not exist
492 /// * `OlderTimestamp` - Timestamp is not newer than existing record
493 ///
494 /// # Example
495 ///
496 /// ```rust
497 /// # use feoxdb::FeoxStore;
498 /// # fn main() -> feoxdb::Result<()> {
499 /// # let store = FeoxStore::new(None)?;
500 /// # store.insert(b"temp", b"data")?;
501 /// store.delete(b"temp")?;
502 /// # Ok(())
503 /// # }
504 /// ```
505 ///
506 /// # Performance
507 ///
508 /// * Memory mode: ~300ns
509 /// * Persistent mode: ~400ns
510 pub fn delete(&self, key: &[u8]) -> Result<()> {
511 self.delete_with_timestamp(key, None)
512 }
513
514 /// Delete a key-value pair with explicit timestamp.
515 ///
516 /// This is the advanced version that allows manual timestamp control.
517 /// Most users should use `delete()` instead.
518 ///
519 /// # Arguments
520 ///
521 /// * `key` - The key to delete
522 /// * `timestamp` - Optional timestamp. If `None`, uses current time.
523 ///
524 /// # Errors
525 ///
526 /// * `OlderTimestamp` - Timestamp is not newer than existing record
527 pub fn delete_with_timestamp(&self, key: &[u8], timestamp: Option<u64>) -> Result<()> {
528 let start = std::time::Instant::now();
529 let timestamp = match timestamp {
530 Some(0) | None => self.get_timestamp(),
531 Some(ts) => ts,
532 };
533 self.validate_key(key)?;
534
535 // Remove from hash table and get the record
536 let record_pair = self.hash_table.remove(key).ok_or(FeoxError::KeyNotFound)?;
537 let record = record_pair.1;
538
539 if timestamp < record.timestamp {
540 // Put it back if timestamp is older
541 self.hash_table.upsert(key.to_vec(), record);
542 return Err(FeoxError::OlderTimestamp);
543 }
544
545 let record_size = record.calculate_size();
546 let old_value_len = record.value_len;
547
548 // Mark record as deleted by setting refcount to 0
549 record.refcount.store(0, Ordering::Release);
550
551 // Remove from lock-free skip list
552 self.tree.remove(key);
553
554 // Update statistics
555 self.stats.record_count.fetch_sub(1, Ordering::AcqRel);
556 self.stats
557 .memory_usage
558 .fetch_sub(record_size, Ordering::AcqRel);
559
560 // Clear from cache
561 if self.enable_caching {
562 if let Some(ref cache) = self.cache {
563 cache.remove(key);
564 }
565 }
566
567 // Queue deletion for persistence if write buffer exists and not memory-only
568 if !self.memory_only {
569 if let Some(ref wb) = self.write_buffer {
570 wb.add_write(Operation::Delete, record, old_value_len)?;
571 }
572 }
573
574 self.stats.record_delete(start.elapsed().as_nanos() as u64);
575 Ok(())
576 }
577
578 /// Get the size of a value without loading it.
579 ///
580 /// Useful for checking value size before loading large values from disk.
581 ///
582 /// # Arguments
583 ///
584 /// * `key` - The key to check
585 ///
586 /// # Returns
587 ///
588 /// Returns the size in bytes of the value.
589 ///
590 /// # Errors
591 ///
592 /// * `KeyNotFound` - Key does not exist
593 ///
594 /// # Example
595 ///
596 /// ```rust
597 /// # use feoxdb::FeoxStore;
598 /// # fn main() -> feoxdb::Result<()> {
599 /// # let store = FeoxStore::new(None)?;
600 /// store.insert(b"large_file", &vec![0u8; 1_000_000])?;
601 ///
602 /// // Check size before loading
603 /// let size = store.get_size(b"large_file")?;
604 /// assert_eq!(size, 1_000_000);
605 /// # Ok(())
606 /// # }
607 /// ```
608 pub fn get_size(&self, key: &[u8]) -> Result<usize> {
609 self.validate_key(key)?;
610
611 let record = self
612 .hash_table
613 .read(key, |_, v| v.clone())
614 .ok_or(FeoxError::KeyNotFound)?;
615
616 Ok(record.value_len)
617 }
618
619 // Internal helper methods
620
621 pub(super) fn validate_key_value(&self, key: &[u8], value: &[u8]) -> Result<()> {
622 if key.is_empty() || key.len() > MAX_KEY_SIZE {
623 return Err(FeoxError::InvalidKeySize);
624 }
625
626 if value.is_empty() || value.len() > MAX_VALUE_SIZE {
627 return Err(FeoxError::InvalidValueSize);
628 }
629
630 Ok(())
631 }
632
633 pub(super) fn validate_key(&self, key: &[u8]) -> Result<()> {
634 if key.is_empty() || key.len() > MAX_KEY_SIZE {
635 return Err(FeoxError::InvalidKeySize);
636 }
637
638 Ok(())
639 }
640
641 pub(super) fn check_memory_limit(&self, size: usize) -> bool {
642 match self.max_memory {
643 Some(limit) => {
644 let current = self.stats.memory_usage.load(Ordering::Acquire);
645 current + size <= limit
646 }
647 None => true,
648 }
649 }
650
651 pub(super) fn calculate_record_size(&self, key_len: usize, value_len: usize) -> usize {
652 std::mem::size_of::<Record>() + key_len + value_len
653 }
654
655 pub(super) fn get_timestamp(&self) -> u64 {
656 self.get_timestamp_pub()
657 }
658}