1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
//! Integrity checking and health monitoring methods for SSTableReader
//!
//! This module contains methods for checking SSTable integrity, monitoring health,
//! and handling tombstone filtering.
use super::super::verify::{self, VerifyMode};
use super::{IntegrityCheckResult, IntegrityStatus, SSTableReader, SSTableReaderHealthMetrics};
use crate::types::{ScanRow, Value};
use crate::Result;
use tracing::{debug, info};
#[cfg(feature = "tombstones")]
use super::super::tombstone_merger::GenerationValue;
#[cfg(feature = "tombstones")]
use crate::{types::TableId, RowKey};
#[cfg(feature = "tombstones")]
use tracing::warn;
impl SSTableReader {
/// Get comprehensive reader health and performance metrics
pub async fn get_health_metrics(&self) -> Result<SSTableReaderHealthMetrics> {
let stats = self.stats().await?;
// Cache health is sourced from the shared B1 decompressed-chunk cache
// (issue #1568): the dead per-reader block cache and its always-empty
// memory summation were deleted, so these numbers now reflect the real
// cache rather than a structural zero.
let cache = self.chunk_cache();
let (hits, misses) = (cache.hit_count(), cache.miss_count());
let cache_hit_rate = if hits + misses > 0 {
hits as f64 / (hits + misses) as f64
} else {
0.0
};
let memory_usage = std::mem::size_of::<Self>() + cache.resident_bytes();
let file_path = self.file_path();
Ok(SSTableReaderHealthMetrics {
file_accessible: file_path.exists(),
file_path,
header_version: self.header.cassandra_version,
total_file_size: stats.file_size,
estimated_memory_usage: memory_usage,
block_cache_entries: cache.len(),
block_cache_hit_rate: cache_hit_rate,
compression_enabled: self.compression_reader.is_some(),
compression_algorithm: self.header.compression.algorithm.clone(),
bloom_filter_enabled: self.bloom_filter.is_some(),
// Issue #2385: the raw-key `index_reader` is THE partition index for a
// BIG SSTable now that the redundant `self.index` build is retired; the
// integrated `self.index` only ever populates for the (inert) in-Data.db
// format. Either being present means index-based lookups are available.
index_available: self.index_reader.is_some() || self.index.is_some(),
generation: self.generation,
last_error: None,
})
}
/// Perform an integrity check on the SSTable.
///
/// Issue #1283: this is a THIN PROJECTION over `verify::verify_sstable` — the
/// single source of truth for SSTable integrity — not an independent check
/// pipeline. The legacy implementation walked only `Data.db` blocks, so a
/// corrupt `Index.db` / `Digest.crc32` / `Summary.db` / `Filter.db` or
/// out-of-order keys (all of which the verifier FAILs) read back `Healthy`
/// here — a divergent verdict. We now run the authoritative verifier in
/// `Full` mode over the reader's EXACT generation (`self.file_path`, not merely
/// its parent directory — roborev #1283) and map its `VerifyReport` onto the
/// legacy `IntegrityCheckResult` shape the (test-only) consumers expect.
pub async fn perform_integrity_check(&self) -> Result<IntegrityCheckResult> {
let file_path = self.file_path();
debug!("Starting integrity check for {:?}", file_path);
// Delegate to the authoritative engine, verifying the EXACT generation this
// reader is opened on (issue #1283, roborev). The directory may hold several
// generations; `verify_sstable` resolves the lexicographically-first
// `*-Data.db`, which would report the wrong SSTable's integrity here.
// `verify_sstable_generation` verifies precisely `self.file_path`'s
// generation. We use the SAME Config/Platform the reader was opened with.
// Data corruption is reported as findings inside an Ok(report); only
// environmental problems return Err.
let report = verify::verify_sstable_generation(
&file_path,
VerifyMode::Full,
&self.open_config,
self.platform.clone(),
)
.await?;
// Project VerifyReport -> IntegrityCheckResult.
// - any finding => Corrupted; none => Healthy (no Degraded — issue #1283).
// - rows_scanned => total_entries.
// - findings' rendered strings => parsing_errors.
// - per-block indices (corrupted_blocks/unreadable_blocks/total_blocks_checked)
// are not produced by the verifier and no production consumer reads them,
// so they stay best-effort empty/zero.
let parsing_errors: Vec<String> = report.findings.iter().map(|f| f.to_string()).collect();
let overall_status = if report.findings.is_empty() {
IntegrityStatus::Healthy
} else {
IntegrityStatus::Corrupted
};
// `checksum_mismatches` is a deprecated, always-0 compatibility field
// (issue #1283): the projection never populates it. The narrow allow keeps
// clippy `-D warnings` clean while the dead computation stays removed.
#[allow(deprecated)]
let result = IntegrityCheckResult {
file_path: file_path.clone(),
total_blocks_checked: 0,
corrupted_blocks: Vec::new(),
checksum_mismatches: 0,
unreadable_blocks: 0,
total_entries: report.rows_scanned.unwrap_or(0),
parsing_errors,
overall_status,
};
info!(
"Integrity check completed for {:?}: {:?}, {} findings, {} rows scanned",
file_path,
result.overall_status,
result.parsing_errors.len(),
result.total_entries
);
Ok(result)
}
/// Enhanced tombstone filtering using TombstoneMerger
#[cfg(feature = "tombstones")]
pub(super) fn filter_tombstone(&self, row: &ScanRow) -> bool {
// Issue #1334: a live row (`ScanRow::Row`) is always kept here — row-level
// tombstone suppression only applies to markers. (Cell tombstones inside a
// live row are preserved for callers to inspect.) Only a marker carries a
// `Value` whose tombstone/TTL semantics this filter evaluates.
let value = match row {
// A live row (decoded or raw undecoded fallback) is always kept.
ScanRow::Row(_) | ScanRow::RawRow(_) => return true,
ScanRow::Marker(v) => v,
};
// Use the fast tombstone check for performance
let write_time = self.extract_write_time_from_value(value);
if self
.tombstone_merger
.fast_tombstone_check(value, write_time)
{
// Value is deleted by tombstone
return false;
}
// Check for TTL expiration on regular values
if let Some(ttl) = self.extract_ttl_from_value(value) {
let current_time = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or_else(|e| {
warn!("Failed to get system time: {}; using fallback value 0", e);
0
});
if current_time > write_time + ttl {
// Value has expired
return false;
}
}
true // Keep valid, non-deleted values
}
/// Simple tombstone filtering (fallback when tombstones feature is disabled).
///
/// Row tombstones (`Value::Tombstone(RowTombstone)`) are always filtered out of
/// user-facing scan/get results, regardless of the `tombstones` feature flag.
/// This prevents deleted rows that are still present on disk (either from a live
/// SSTable that contains a tombstone entry, or from a post-compaction SSTable
/// that preserved tombstone rows for GC purposes) from appearing in query results.
///
/// Cell tombstones (`Value::Tombstone(CellTombstone)`) within a Map are NOT
/// filtered here — they are preserved so callers can inspect them. If a caller
/// needs to suppress null-cell entries, it should do so at the query layer.
///
/// (Issue #505)
#[cfg(not(feature = "tombstones"))]
pub(super) fn filter_tombstone(&self, row: &ScanRow) -> bool {
use crate::types::TombstoneType;
// Issue #1334: a live row (`ScanRow::Row`) is always kept; filter out only a
// row-level tombstone marker.
!matches!(
row,
ScanRow::Marker(Value::Tombstone(info))
if info.tombstone_type == TombstoneType::RowTombstone
)
}
/// Enhanced multi-generation tombstone filtering for compaction
#[cfg(feature = "tombstones")]
pub async fn filter_with_multi_generation_merge(
&self,
table_id: &TableId,
entries: Vec<(RowKey, Vec<GenerationValue>)>,
) -> Result<Vec<(RowKey, ScanRow)>> {
let mut results = Vec::new();
tracing::debug!(
"Processing {} key groups for multi-generation merge",
entries.len()
);
// Use batch processing for better performance
const BATCH_SIZE: usize = 1000;
let batches: Vec<_> = entries.chunks(BATCH_SIZE).collect();
for (batch_idx, batch) in batches.iter().enumerate() {
tracing::debug!(
"Processing batch {}/{} with {} entries",
batch_idx + 1,
batches.len(),
batch.len()
);
let batch_entries = batch.to_vec();
let merged_results = self
.tombstone_merger
.batch_merge_with_tombstones(batch_entries, BATCH_SIZE)?;
for (key, merged_value) in merged_results {
if let Some(value) = merged_value {
if self.should_include_value_after_merge(&value, table_id, &key)? {
results.push((key, value));
}
} else {
// Value was completely tombstoned
tracing::debug!("Value for key {:?} was completely tombstoned", key);
}
}
}
tracing::debug!(
"Multi-generation merge completed: {} final results from {} input groups",
results.len(),
entries.len()
);
Ok(results)
}
/// Enhanced filtering logic for post-merge values including collection validation
#[cfg(feature = "tombstones")]
fn should_include_value_after_merge(
&self,
row: &ScanRow,
_table_id: &TableId,
_key: &RowKey,
) -> Result<bool> {
// Issue #1334: the merge now yields whole rows. A live row with at least one
// cell is included; a marker (row tombstone / null row) or an empty row is
// suppressed.
match row {
ScanRow::Row(cells) => Ok(!cells.is_empty()),
// A raw undecoded fallback row carries live bytes → included.
ScanRow::RawRow(bytes) => Ok(!bytes.is_empty()),
ScanRow::Marker(_) => Ok(false),
}
}
/// Extract TTL from value metadata
#[cfg(feature = "tombstones")]
fn extract_ttl_from_value(&self, value: &Value) -> Option<i64> {
match value {
Value::Tombstone(info) => info.ttl,
_ => None, // Regular values would have TTL in SSTable metadata
}
}
/// Extract write time from value
#[cfg(feature = "tombstones")]
fn extract_write_time_from_value(&self, value: &Value) -> i64 {
match value {
Value::Tombstone(info) => info.deletion_time,
_ => std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or_else(|e| {
warn!("Failed to get system time: {}; using fallback value 0", e);
0
}),
}
}
}