diff --git a/native/src/txlog/arrow_ffi.rs b/native/src/txlog/arrow_ffi.rs index 6755e233..a7d1876d 100644 --- a/native/src/txlog/arrow_ffi.rs +++ b/native/src/txlog/arrow_ffi.rs @@ -18,6 +18,24 @@ use arrow::record_batch::RecordBatch; use super::actions::FileEntry; use super::error::{TxLogError, Result}; +/// Convert a time-range stat value to epoch milliseconds for the Int64 Arrow column. +/// +/// `AddAction::time_range_{start,end}` is stored as a String that may be either the +/// legacy integer form (already epoch millis) or the Scala ISO-8601 form. Parsing +/// with `i64::from_str` alone drops ISO-8601 values to null, losing time-range +/// pruning data for Scala-written logs (TXLOG_MODULE_REVIEW D6). +/// +/// The datetime-string formats (RFC 3339, bare datetime, date-only) are parsed by +/// the shared `partition_pruning::parse_datetime_string_to_micros` so this and the +/// stats-comparison path accept exactly the same set of formats. +fn time_range_to_epoch_millis(s: &str) -> Option { + if let Ok(v) = s.parse::() { + return Some(v); // legacy epoch millis + } + // Shared parser returns micros; scale to the Int64 column's millisecond unit. + super::partition_pruning::parse_datetime_string_to_micros(s).map(|micros| micros / 1_000) +} + /// Build the Arrow schema dynamically based on partition columns and options. /// /// Base schema has 20 fixed columns. Additional columns are appended: @@ -129,8 +147,8 @@ pub fn file_entries_to_record_batch( delete_opstamp_builder.append_option(add.delete_opstamp); num_merge_ops_builder.append_option(add.num_merge_ops); uncomp_size_builder.append_option(add.uncompressed_size_bytes); - time_start_builder.append_option(add.time_range_start.as_ref().and_then(|s| s.parse::().ok())); - time_end_builder.append_option(add.time_range_end.as_ref().and_then(|s| s.parse::().ok())); + time_start_builder.append_option(add.time_range_start.as_deref().and_then(time_range_to_epoch_millis)); + time_end_builder.append_option(add.time_range_end.as_deref().and_then(time_range_to_epoch_millis)); comp_delta_ver_builder.append_option(add.companion_delta_version); match &add.doc_mapping_json { diff --git a/native/src/txlog/cache.rs b/native/src/txlog/cache.rs index 77c64800..6f7d2c4a 100644 --- a/native/src/txlog/cache.rs +++ b/native/src/txlog/cache.rs @@ -109,10 +109,10 @@ impl CacheConfig { cfg.file_list_capacity = n; } - // Concurrency - if let Some(n) = map.get("max_concurrent_reads").and_then(|s| s.parse::().ok()) { - cfg.max_concurrent_reads = if n == 0 { 32 } else { n }; - } + // Concurrency — single source of truth for parsing this key, shared with + // the direct callers in distributed.rs (prevents the default/semantics from + // drifting between the two parse sites; see TXLOG_MODULE_REVIEW E4). + cfg.max_concurrent_reads = super::distributed::extract_max_concurrent(map); // Kill-switch if map.get("cache.enabled").map(|s| s.as_str()) == Some("false") { @@ -502,6 +502,23 @@ pub fn get_or_create_cache(table_path: &str, config: CacheConfig) -> Arc). Once past the soft cap, drop + // entries no longer referenced by any live caller — `strong_count == 1` means only + // the registry holds it, so evicting is safe (it is recreated on the next access). + // + // This is a best-effort reclaim of *idle* tables, NOT a hard cap: if more than + // MAX_REGISTRY_ENTRIES tables are being queried concurrently, every entry has + // strong_count > 1 and nothing is freed. Bounding that case would require evicting + // live entries (causing duplicate caches for the same table) or a real LRU; the + // idle-reclaim covers the realistic "many tables over a long lifetime" pattern. + // See TXLOG_MODULE_REVIEW E4. + const MAX_REGISTRY_ENTRIES: usize = 256; + if registry.len() >= MAX_REGISTRY_ENTRIES { + registry.retain(|_, cache| Arc::strong_count(cache) > 1); + } + let cache = Arc::new(TxLogCache::new(config)); registry.insert(key, cache.clone()); cache diff --git a/native/src/txlog/distributed.rs b/native/src/txlog/distributed.rs index e26c5271..0cbf6039 100644 --- a/native/src/txlog/distributed.rs +++ b/native/src/txlog/distributed.rs @@ -54,8 +54,16 @@ pub(crate) fn parse_metadata_json(json_str: &str) -> Option { serde_json::from_value(inner.clone()).ok() } -/// Read multiple version files concurrently, silently ignoring all errors -/// (including NotFound). Returns (version, actions) pairs sorted ascending. +/// Read multiple version files concurrently. Returns (version, actions) pairs +/// sorted ascending. +/// +/// A **NotFound** on any version is tolerated (skipped) — this handles benign +/// TRUNCATE/PURGE races where a version file was legitimately removed. Any other +/// error (transient S3/Azure throttle, network failure, corrupt file) is +/// **propagated**, because silently dropping such a version would produce a +/// snapshot/checkpoint that is missing committed adds or removes. See the sibling +/// `read_post_checkpoint_changes` which uses the same NotFound-skip / else-propagate +/// discipline. /// /// `max_concurrent` bounds the number of in-flight object-store GETs to prevent /// S3/Azure rate-limit throttling on large catch-up reads. @@ -63,24 +71,29 @@ async fn read_versions_concurrent( storage: &TxLogStorage, versions: impl IntoIterator, max_concurrent: usize, -) -> Vec<(i64, Vec)> { +) -> Result)>> { let versions: Vec = versions.into_iter().collect(); let futures = versions.into_iter().map(|v| { let storage = storage; // copy &TxLogStorage (references are Copy) async move { match version_file::read_version(storage, v).await { - Ok(actions) => Some((v, actions)), - Err(_) => None, + Ok(actions) => Ok(Some((v, actions))), + Err(e) if is_not_found_error(&e) => Ok(None), // benign TRUNCATE/PURGE race + Err(e) => Err(e), // transient/real I/O error → propagate } } }); let mut results: Vec<(i64, Vec)> = stream::iter(futures) .buffer_unordered(max_concurrent) - .filter_map(|opt| async move { opt }) - .collect() - .await; + .collect::)>>>>() + .await + .into_iter() + .collect::)>>>>()? + .into_iter() + .flatten() + .collect(); results.sort_by_key(|(v, _)| *v); - results + Ok(results) } /// Run up to `max_concurrent` futures concurrently, collecting results in order. @@ -112,7 +125,14 @@ where /// Returns true if the error represents a "not found" condition from /// any object store backend (S3, Azure, local filesystem). +/// +/// Prefers the structural `TxLogError::NotFound` variant (produced by +/// `TxLogStorage::get`); falls back to substring matching only for errors that +/// still flatten a NotFound into a formatted message from another code path. fn is_not_found_error(e: &TxLogError) -> bool { + if matches!(e, TxLogError::NotFound { .. }) { + return true; + } let s = e.to_string(); s.contains("not found") || s.contains("NotFound") || s.contains("No such file") || s.contains("404") @@ -133,6 +153,12 @@ pub struct TxLogSnapshotInfo { pub state_dir: String, /// Tombstones from the checkpoint's StateManifest (paths of removed files). pub tombstones: Vec, + /// Post-checkpoint version actions already parsed while building this snapshot + /// (for protocol/metadata override). When `Some`, `list_files` reuses these + /// instead of re-fetching the same version files — eliminating a 2× GET per + /// query on tables with post-checkpoint commits (TXLOG_MODULE_REVIEW E1). + /// `None` means "not captured; fall back to reading `post_checkpoint_version_paths`". + pub post_checkpoint_actions: Option)>>, } #[derive(Debug, Clone)] @@ -213,19 +239,23 @@ pub async fn get_txlog_snapshot_info_with_cache( // If there are post-checkpoint versions, read them for potential // protocol/metadata overrides. This is critical for concurrent - // metadata updates — the cached metadata may be stale. + // metadata updates — the cached metadata may be stale. The parsed + // actions are captured and reused by list_files (E1: no re-fetch). + let mut captured_actions: Option)>> = None; let (effective_protocol, mut effective_metadata) = if post_cp_paths.is_empty() { (Some(cached_meta.0.clone()), cached_meta.1.clone()) } else { let post_cp_versions: Vec = all_versions.iter().copied() .filter(|&v| v > cached_cp.version) .collect(); - let post_cp_actions = read_versions_concurrent(&storage, post_cp_versions, max_concurrent).await; + let post_cp_actions = read_versions_concurrent(&storage, post_cp_versions, max_concurrent).await?; let (post_protocol, post_metadata) = log_replay::extract_metadata(&[], &post_cp_actions); - ( + let result = ( post_protocol.or(Some(cached_meta.0.clone())), post_metadata.unwrap_or_else(|| cached_meta.1.clone()), - ) + ); + captured_actions = Some(post_cp_actions); + result }; // Bug 2b fix: Merge schema_registry into effective_metadata.configuration @@ -245,6 +275,7 @@ pub async fn get_txlog_snapshot_info_with_cache( metadata: effective_metadata, state_dir, tombstones: state_manifest.tombstones.clone(), + post_checkpoint_actions: captured_actions, }); } } @@ -302,7 +333,7 @@ async fn snapshot_from_checkpoint( let post_cp_paths: Vec = post_cp_versions.iter() .map(|v| TxLogStorage::version_path(*v)) .collect(); - let all_version_actions = read_versions_concurrent(storage, post_cp_versions.clone(), max_concurrent).await; + let all_version_actions = read_versions_concurrent(storage, post_cp_versions.clone(), max_concurrent).await?; // Source protocol/metadata from the checkpoint's cached JSON first, // then override with any updates from post-checkpoint version files. @@ -431,6 +462,8 @@ async fn snapshot_from_checkpoint( metadata, state_dir, tombstones: state_manifest.tombstones.clone(), + // Reuse the already-parsed post-checkpoint actions in list_files (E1). + post_checkpoint_actions: Some(all_version_actions), }) } @@ -451,7 +484,7 @@ async fn snapshot_from_version_scan( } // Read ALL version files to extract protocol/metadata — concurrently - let all_version_actions = read_versions_concurrent(storage, all_versions.iter().copied(), max_concurrent).await; + let all_version_actions = read_versions_concurrent(storage, all_versions.iter().copied(), max_concurrent).await?; // Extract protocol/metadata from all versions let v0_actions = all_version_actions.iter() @@ -488,6 +521,8 @@ async fn snapshot_from_version_scan( metadata, state_dir: String::new(), tombstones: vec![], + // All version files were already read here; reuse them in list_files (E1). + post_checkpoint_actions: Some(all_version_actions), }) } @@ -608,6 +643,40 @@ pub async fn read_post_checkpoint_changes( }) } +/// Build `TxLogChanges` from post-checkpoint version actions that were already +/// parsed while constructing the snapshot (E1) — avoids re-fetching the same +/// version files in `list_files`. Mirrors the action processing in +/// `read_post_checkpoint_changes`. +pub fn changes_from_versioned_actions( + versioned: &[(i64, Vec)], + metadata_config: &HashMap, +) -> TxLogChanges { + let max_version = versioned.iter().map(|(v, _)| *v).max().unwrap_or(0); + let mut added = Vec::new(); + let mut removed = Vec::new(); + let mut skips = Vec::new(); + for (version, actions) in versioned { + for action in actions { + match action { + Action::Add(add) => { + let mut add = add.clone(); + schema_dedup::restore_schemas_on_adds(std::slice::from_mut(&mut add), metadata_config); + let timestamp = add.modification_time; + added.push(FileEntry { + add, + added_at_version: *version, + added_at_timestamp: timestamp, + }); + } + Action::Remove(r) => removed.push(r.path.clone()), + Action::MergeSkip(s) => skips.push(s.clone()), + _ => {} + } + } + } + TxLogChanges { added_files: added, removed_paths: removed, skip_actions: skips, max_version } +} + // ============================================================================ // Executor-side primitive: read_manifest // ============================================================================ @@ -674,6 +743,36 @@ pub struct WriteResult { pub conflicted_versions: Vec, } +/// Flag (via debug log) any path that is both Added and Removed in the same batch +/// of actions — a violation of the never-re-added invariant that would break the +/// unordered add/remove merge in list_files. O(n), no I/O. +/// +/// The whole scan is a diagnostic whose only output is a debug log, so it is skipped +/// entirely unless debug logging is enabled — no HashSet allocation or scan on the +/// hot write path in production (TANTIVY4JAVA_DEBUG off). +fn detect_readd_within_batch(actions: &[Action]) { + if !*crate::debug::DEBUG_ENABLED { + return; + } + let added: std::collections::HashSet<&str> = actions.iter() + .filter_map(|a| match a { Action::Add(add) => Some(add.path.as_str()), _ => None }) + .collect(); + if added.is_empty() { + return; + } + for a in actions { + if let Action::Remove(r) = a { + if added.contains(r.path.as_str()) { + debug_println!( + "⚠️ DISTRIBUTED: never-re-added invariant violated — path is both Added and Removed \ + in the same version batch: {}. Unordered add/remove merging assumes this never happens.", + r.path + ); + } + } + } +} + /// Write a new version file with automatic conflict retry. pub async fn write_version( table_path: &str, @@ -682,6 +781,14 @@ pub async fn write_version( retry_config: RetryConfig, ) -> Result { let storage = TxLogStorage::new(table_path, config)?; + + // Cheap enforcement of the never-re-added invariant (TXLOG_MODULE_REVIEW C1): + // detect a path that is both Added and Removed within this same batch. A true + // cross-version check would require reading the current tombstone set on every + // write (not cheap), so we only flag the free, in-batch contradiction here. The + // unordered add/remove merge in list_files relies on this never happening. + detect_readd_within_batch(&actions); + let mut conflicted = Vec::new(); // On the first attempt we LIST to find the current version. On each conflict // the correct next candidate is last_conflict + 1 — no re-LIST needed. @@ -721,11 +828,15 @@ pub async fn write_version( retry_config.base_delay_ms * (1 << attempt.min(10)), retry_config.max_delay_ms, ); - // Add ±25% jitter + // Add ±25% jitter. Mix the process id into the seed so contending processes + // at the same attempt number get *different* delays — without it, every + // process computes an identical jitter and the thundering herd is not + // decorrelated at all. (No `rand` dependency needed.) let jitter_range = base_delay / 4; let jitter = if jitter_range > 0 { - // Simple deterministic jitter based on attempt number to avoid rand dependency - (attempt as u64 * 7919) % (jitter_range * 2) // pseudo-random spread + let seed = (attempt as u64).wrapping_mul(7919) + .wrapping_add((std::process::id() as u64).wrapping_mul(2_654_435_761)); + seed % (jitter_range * 2) // pseudo-random spread, per-process } else { 0 }; @@ -817,24 +928,36 @@ pub async fn initialize_table( // Auto-checkpoint support (GAP-10) // ============================================================================ -/// Check if auto-checkpoint is enabled. Disabled when checkpoint_interval=0. -fn is_auto_checkpoint_enabled(config_map: &HashMap) -> bool { - let val_str = config_map.get("checkpoint_interval") - .or_else(|| config_map.get("checkpoint.interval")); - match val_str { - Some(s) => { - let interval: i64 = s.parse().unwrap_or(1); - interval > 0 - } - None => true, // enabled by default - } +/// Resolve the configured checkpoint interval. +/// +/// - Missing / unparseable → 1 (checkpoint after every commit — the default). +/// - `0` (or negative) → 0, which disables auto-checkpointing entirely. +/// - `N > 0` → checkpoint only on versions where `version % N == 0`. +fn checkpoint_interval(config_map: &HashMap) -> i64 { + config_map.get("checkpoint_interval") + .or_else(|| config_map.get("checkpoint.interval")) + .and_then(|s| s.parse::().ok()) + .map(|n| n.max(0)) + .unwrap_or(1) +} + +/// Whether a checkpoint should be written at `written_version`, honoring the +/// configured interval. Interval `N` means "checkpoint every N commits", so a +/// checkpoint is only written when `written_version` is a multiple of `N`. +/// Interval `0` disables checkpointing. +fn should_checkpoint_at(config_map: &HashMap, written_version: i64) -> bool { + let interval = checkpoint_interval(config_map); + interval > 0 && written_version >= 0 && written_version % interval == 0 } -/// Create a checkpoint (state directory + _last_checkpoint) after every write. +/// Create a checkpoint (state directory + _last_checkpoint) when `written_version` +/// lands on a checkpoint boundary — i.e. every `checkpoint_interval` commits +/// (`written_version % interval == 0`; default interval 1 = every write). See +/// `should_checkpoint_at`. /// -/// Matches Scala behavior: uses incremental writes when a previous checkpoint exists -/// (reusing existing manifest references and accumulating tombstones), falling back to -/// compacted writes when no checkpoint exists or tombstone ratio is high. +/// Uses incremental writes when a previous checkpoint exists (reusing existing +/// manifest references and accumulating tombstones), falling back to compacted writes +/// when no checkpoint exists or the tombstone ratio is high. /// Set checkpoint_interval=0 to disable entirely. /// Failures are logged but never fail the write operation. pub async fn maybe_auto_checkpoint( @@ -843,7 +966,7 @@ pub async fn maybe_auto_checkpoint( config_map: &HashMap, written_version: i64, ) { - if written_version < 0 || !is_auto_checkpoint_enabled(config_map) { + if !should_checkpoint_at(config_map, written_version) { return; } let max_concurrent = extract_max_concurrent(config_map); @@ -877,7 +1000,16 @@ pub async fn maybe_auto_checkpoint( let post_cp_versions: Vec = all_versions.iter().copied() .filter(|&v| v > cp_version && v <= written_version) .collect(); - let post_cp_actions = read_versions_concurrent(&storage, post_cp_versions, max_concurrent).await; + // Abort (rather than persist a lossy checkpoint) if any version read + // fails with a non-NotFound error. Auto-checkpoint is best-effort; a + // transient read error here must not drop committed adds/removes. + let post_cp_actions = match read_versions_concurrent(&storage, post_cp_versions, max_concurrent).await { + Ok(a) => a, + Err(e) => { + debug_println!("⚠️ DISTRIBUTED: auto-checkpoint aborted, read_versions_concurrent failed: {}", e); + return; + } + }; // Extract new adds and removes from post-checkpoint versions let mut new_adds: Vec = Vec::new(); @@ -910,129 +1042,52 @@ pub async fn maybe_auto_checkpoint( use super::tombstone_distributor; if tombstone_distributor::needs_compaction(&base_manifest, removed_paths.len()) { - // Compaction needed — try selective first, fall back to full - let partition_columns = metadata.partition_columns.clone(); - let all_tombstones: std::collections::HashSet = base_manifest.tombstones.iter() - .chain(removed_paths.iter()) - .cloned() + // Full compaction — read every manifest, apply tombstones, replay, write fresh. + // + // NOTE: a "selective" compaction path previously lived here but read the + // dirty manifests AND all clean manifests and then wrote a fully compacted + // checkpoint anyway — producing an output identical to full compaction while + // adding a second manifest fan-out and a path-based partition heuristic. It + // was collapsed into full compaction (TXLOG_MODULE_REVIEW E3). A genuine + // selective compaction would reuse kept manifest refs (via + // write_incremental_state_directory) rather than re-reading and rewriting + // them; until that exists, one code path is correct and cheaper. + debug_println!("📊 DISTRIBUTED: full compaction ({} base tombstones + {} new removes)", + base_manifest.tombstones.len(), removed_paths.len()); + let state_dir = cp_info.state_dir.unwrap_or_else(|| TxLogStorage::state_dir_name(cp_version)); + let metadata_config = base_manifest.schema_registry.clone(); + // Read all checkpoint manifests concurrently (bounded). + let cp_futs: Vec<_> = base_manifest.manifests.iter() + .map(|mi| super::avro::state_reader::read_single_manifest( + &storage, &state_dir, &mi.path, &metadata_config, + )) .collect(); - - let manifests_with_tombstones = tombstone_distributor::distribute_tombstones_to_manifests( - &base_manifest.manifests, &all_tombstones, &partition_columns, - ); - let (keep, rewrite) = tombstone_distributor::selective_partition( - &manifests_with_tombstones, tombstone_distributor::COMPACTION_TOMBSTONE_THRESHOLD, - ); - - if tombstone_distributor::is_selective_compaction_beneficial(&keep, &rewrite) { - debug_println!("📊 DISTRIBUTED: selective compaction: keeping {} clean, rewriting {} dirty manifests", - keep.len(), rewrite.len()); - - // Read only dirty manifests, filter tombstones, write new - let state_dir = cp_info.state_dir.clone() - .unwrap_or_else(|| TxLogStorage::state_dir_name(cp_version)); - let metadata_config = base_manifest.schema_registry.clone(); - // Read dirty manifests concurrently (bounded); filter tombstones after. - let rewrite_futs: Vec<_> = rewrite.iter() - .map(|mi| super::avro::state_reader::read_single_manifest( - &storage, &state_dir, &mi.path, &metadata_config, - )) - .collect(); - let rewrite_results = stream::iter(rewrite_futs) - .buffer_unordered(max_concurrent) - .collect::>>>().await; - let rewritten_entries: Vec = match rewrite_results - .into_iter() - .collect::>>>() - { - Err(e) => { - debug_println!("⚠️ DISTRIBUTED: selective compaction aborted — dirty manifest read failed: {}", e); - return; - } - Ok(nested) => nested.into_iter() - .flat_map(|entries| entries.into_iter() - .filter(|e| !all_tombstones.contains(&e.add.path))) - .collect(), - }; - // Combine: new adds (filtered against removes) + rewritten live entries - let mut all_live = rewritten_entries; - all_live.extend(new_adds.iter() - .filter(|e| !all_tombstones.contains(&e.add.path)) - .cloned()); - - // Clean manifests are reused; write new manifest for rewritten + new - // For simplicity, do a compacted write with all live files from - // kept manifests + rewritten entries + new adds. - // A full selective compaction would reuse kept manifest refs, but - // that requires state_writer changes beyond scope here. - // Instead, read kept manifest entries too for a clean compacted write. - // Read clean manifests concurrently (bounded). - let keep_futs: Vec<_> = keep.iter() - .map(|mi| super::avro::state_reader::read_single_manifest( - &storage, &state_dir, &mi.path, &metadata_config, - )) - .collect(); - let keep_results = stream::iter(keep_futs) - .buffer_unordered(max_concurrent) - .collect::>>>().await; - let keep_live: Vec = match keep_results - .into_iter() - .collect::>>>() - { - Err(e) => { - debug_println!("⚠️ DISTRIBUTED: selective compaction aborted — clean manifest read failed: {}", e); - return; - } - Ok(nested) => nested.into_iter() - .flat_map(|entries| entries.into_iter() - .filter(|e| !all_tombstones.contains(&e.add.path))) - .collect(), - }; - all_live.extend(keep_live); - all_live.sort_by(|a, b| a.add.path.cmp(&b.add.path)); - - match write_checkpoint_at_version(table_path, config, all_live, metadata, protocol, written_version).await { - Ok(info) => debug_println!("✅ DISTRIBUTED: selective compaction at v{}, {} files", info.version, info.num_files), - Err(e) => debug_println!("⚠️ DISTRIBUTED: selective compaction failed (non-fatal): {}", e), - } - } else { - // Full compaction — read everything, apply tombstones, replay, write fresh - debug_println!("📊 DISTRIBUTED: full compaction (selective not beneficial)"); - let state_dir = cp_info.state_dir.unwrap_or_else(|| TxLogStorage::state_dir_name(cp_version)); - let metadata_config = base_manifest.schema_registry.clone(); - // Read all checkpoint manifests concurrently (bounded). - let cp_futs: Vec<_> = base_manifest.manifests.iter() - .map(|mi| super::avro::state_reader::read_single_manifest( - &storage, &state_dir, &mi.path, &metadata_config, - )) - .collect(); - let cp_results = stream::iter(cp_futs) - .buffer_unordered(max_concurrent) - .collect::>>>().await; - let mut cp_entries: Vec = match cp_results - .into_iter() - .collect::>>>() - { - Err(e) => { - debug_println!("⚠️ DISTRIBUTED: full compaction aborted — manifest read failed: {}", e); - return; - } - Ok(nested) => nested.into_iter().flatten().collect(), - }; - // Filter out tombstoned entries from base checkpoint before replay - if !base_manifest.tombstones.is_empty() { - let tombstone_set: std::collections::HashSet<&str> = - base_manifest.tombstones.iter().map(|s| s.as_str()).collect(); - let before = cp_entries.len(); - cp_entries.retain(|e| !tombstone_set.contains(e.add.path.as_str())); - debug_println!("📖 DISTRIBUTED: Applied {} base tombstones during compaction, {} → {} entries", - base_manifest.tombstones.len(), before, cp_entries.len()); - } - let replay_result = log_replay::replay(cp_entries, post_cp_actions); - match write_checkpoint_at_version(table_path, config, replay_result.files, metadata, protocol, written_version).await { - Ok(info) => debug_println!("✅ DISTRIBUTED: full compaction at v{}, {} files", info.version, info.num_files), - Err(e) => debug_println!("⚠️ DISTRIBUTED: full compaction failed (non-fatal): {}", e), + let cp_results = stream::iter(cp_futs) + .buffer_unordered(max_concurrent) + .collect::>>>().await; + let mut cp_entries: Vec = match cp_results + .into_iter() + .collect::>>>() + { + Err(e) => { + debug_println!("⚠️ DISTRIBUTED: full compaction aborted — manifest read failed: {}", e); + return; } + Ok(nested) => nested.into_iter().flatten().collect(), + }; + // Filter out tombstoned entries from base checkpoint before replay + if !base_manifest.tombstones.is_empty() { + let tombstone_set: std::collections::HashSet<&str> = + base_manifest.tombstones.iter().map(|s| s.as_str()).collect(); + let before = cp_entries.len(); + cp_entries.retain(|e| !tombstone_set.contains(e.add.path.as_str())); + debug_println!("📖 DISTRIBUTED: Applied {} base tombstones during compaction, {} → {} entries", + base_manifest.tombstones.len(), before, cp_entries.len()); + } + let replay_result = log_replay::replay(cp_entries, post_cp_actions); + match write_checkpoint_at_version(table_path, config, replay_result.files, metadata, protocol, written_version).await { + Ok(info) => debug_println!("✅ DISTRIBUTED: full compaction at v{}, {} files", info.version, info.num_files), + Err(e) => debug_println!("⚠️ DISTRIBUTED: full compaction failed (non-fatal): {}", e), } } else { // Low tombstone ratio — incremental write (reuse existing manifests) @@ -1060,7 +1115,14 @@ pub async fn maybe_auto_checkpoint( let versions_to_read: Vec = all_versions.iter().copied() .filter(|&v| v <= written_version) .collect(); - let versioned_actions = read_versions_concurrent(&storage, versions_to_read, max_concurrent).await; + // Abort rather than persist a lossy checkpoint on transient read errors. + let versioned_actions = match read_versions_concurrent(&storage, versions_to_read, max_concurrent).await { + Ok(a) => a, + Err(e) => { + debug_println!("⚠️ DISTRIBUTED: auto-checkpoint aborted, read_versions_concurrent failed: {}", e); + return; + } + }; let v0_actions = versioned_actions.iter() .find(|(v, _)| *v == 0) .map(|(_, a)| a.clone()) @@ -1091,8 +1153,13 @@ pub async fn write_checkpoint( protocol: ProtocolAction, ) -> Result { let storage = TxLogStorage::new(table_path, config)?; - let version = storage.list_versions().await? - .last().copied().unwrap_or(0); + // On an uninitialized table there are no version files; writing a checkpoint for + // a phantom version 0 would produce a checkpoint referencing a commit that does + // not exist. Refuse rather than persist that inconsistency. + let version = match storage.list_versions().await?.last().copied() { + Some(v) => v, + None => return Err(TxLogError::NotInitialized { path: table_path.to_string() }), + }; match super::avro::state_writer::write_state_checkpoint( &storage, version, &entries, &protocol, &metadata, @@ -1230,6 +1297,19 @@ pub(crate) async fn probe_versions_since( } } + // Non-contiguous frontier: a version beyond the contiguous run still exists, + // meaning a mid-sequence version file was deleted (operator error / partial + // delete). Contiguity probing would stop at the gap and hide every later + // commit until TTL expiry, so recover them now with a full LIST. The batch + // results already told us a later version exists. + if results.iter().any(|(pv, exists)| *exists && *pv > v) { + debug_println!("⚠️ DISTRIBUTED: probe_versions_since found gap before an existing version at v{}, falling back to list_versions", v); + let all = storage.list_versions().await?; + let remaining: Vec = all.into_iter().filter(|&x| x >= v).collect(); + versions.extend(remaining); + break; + } + if !any_hit { break; // No newer version exists } @@ -1337,8 +1417,77 @@ async fn read_previous_checkpoint( let protocol: Option = manifest.protocol_json.as_ref() .and_then(|s| serde_json::from_str(s).ok()); - let metadata: Option = manifest.metadata.as_ref() + let mut metadata: Option = manifest.metadata.as_ref() .and_then(|s| parse_metadata_json(s)); + // Merge the checkpoint's schema_registry into metadata.configuration so the + // registry survives into incrementally-written checkpoints. Scala-written + // checkpoints keep the registry only in StateManifest.schemaRegistry (not in + // metadata.configuration); without this merge, write_incremental_state_directory + // rebuilds the registry from metadata.configuration alone and silently drops + // those entries, breaking later doc_mapping_ref → doc_mapping_json resolution. + // Mirrors the same merge in snapshot_from_checkpoint(). + if let Some(m) = metadata.as_mut() { + for (k, v) in &manifest.schema_registry { + m.configuration.entry(k.clone()).or_insert_with(|| v.clone()); + } + } + Some((cp_info, manifest, protocol, metadata)) } + +#[cfg(test)] +mod checkpoint_interval_tests { + use super::*; + + fn cfg(pairs: &[(&str, &str)]) -> HashMap { + pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect() + } + + #[test] + fn test_default_interval_checkpoints_every_commit() { + // No config → interval 1 → every version checkpoints. + let c = cfg(&[]); + assert_eq!(checkpoint_interval(&c), 1); + for v in 0..5 { + assert!(should_checkpoint_at(&c, v), "v{} should checkpoint by default", v); + } + } + + #[test] + fn test_zero_disables_checkpointing() { + let c = cfg(&[("checkpoint_interval", "0")]); + assert_eq!(checkpoint_interval(&c), 0); + for v in 0..5 { + assert!(!should_checkpoint_at(&c, v)); + } + } + + #[test] + fn test_interval_is_honored_not_boolean() { + // interval 10 must NOT behave like 1 — only multiples of 10 checkpoint. + let c = cfg(&[("checkpoint_interval", "10")]); + assert_eq!(checkpoint_interval(&c), 10); + assert!(should_checkpoint_at(&c, 0)); + assert!(!should_checkpoint_at(&c, 1)); + assert!(!should_checkpoint_at(&c, 9)); + assert!(should_checkpoint_at(&c, 10)); + assert!(!should_checkpoint_at(&c, 15)); + assert!(should_checkpoint_at(&c, 20)); + } + + #[test] + fn test_negative_version_never_checkpoints() { + let c = cfg(&[("checkpoint_interval", "1")]); + assert!(!should_checkpoint_at(&c, -1)); + } + + #[test] + fn test_dotted_key_alias_and_unparseable_default() { + assert_eq!(checkpoint_interval(&cfg(&[("checkpoint.interval", "5")])), 5); + // Unparseable falls back to the default of 1. + assert_eq!(checkpoint_interval(&cfg(&[("checkpoint_interval", "nonsense")])), 1); + // Negative clamps to 0 (disabled). + assert_eq!(checkpoint_interval(&cfg(&[("checkpoint_interval", "-3")])), 0); + } +} diff --git a/native/src/txlog/error.rs b/native/src/txlog/error.rs index 59ed9f74..b135f9a5 100644 --- a/native/src/txlog/error.rs +++ b/native/src/txlog/error.rs @@ -19,6 +19,9 @@ pub enum TxLogError { #[error("Checkpoint corrupted at version {version}: {detail}")] CorruptedCheckpoint { version: i64, detail: String }, + #[error("Not found: {path}")] + NotFound { path: String }, + #[error("Storage error: {0}")] Storage(anyhow::Error), diff --git a/native/src/txlog/garbage_collection.rs b/native/src/txlog/garbage_collection.rs deleted file mode 100644 index 21562173..00000000 --- a/native/src/txlog/garbage_collection.rs +++ /dev/null @@ -1,250 +0,0 @@ -// txlog/garbage_collection.rs - Clean up old version files and orphaned state directories -// -// Removes version files older than retention period and state directories -// not referenced by the current checkpoint. - -use super::error::Result; -use super::storage::TxLogStorage; - -/// Configuration for garbage collection. -#[derive(Debug, Clone)] -pub struct GcConfig { - /// Number of days to retain version files after they are checkpointed. - pub log_retention_days: u32, - /// Number of hours to retain orphaned state directories. - pub checkpoint_retention_hours: u32, -} - -impl Default for GcConfig { - fn default() -> Self { - Self { - log_retention_days: 30, - checkpoint_retention_hours: 2, - } - } -} - -/// Result of a garbage collection run. -#[derive(Debug, Clone, Default)] -pub struct GcResult { - pub deleted_version_files: u32, - pub deleted_state_dirs: u32, - pub bytes_freed: i64, -} - -/// Clean up old version files and orphaned state directories. -/// -/// Steps: -/// 1. Read `_last_checkpoint` to find the current checkpoint version. -/// 2. List all version files; delete those at or before the checkpoint version -/// AND older than the retention period. -/// 3. List all `state-v/` directories; delete those not referenced by -/// `_last_checkpoint` and older than `checkpoint_retention_hours`. -pub async fn garbage_collect( - storage: &TxLogStorage, - config: &GcConfig, -) -> Result { - let mut result = GcResult::default(); - - // 1. Read _last_checkpoint to get current checkpoint version - let checkpoint_data = storage.get("_last_checkpoint").await?; - let last_cp: super::actions::LastCheckpointInfo = serde_json::from_slice(&checkpoint_data)?; - let checkpoint_version = last_cp.version; - let checkpoint_state_dir = last_cp.state_dir.clone() - .unwrap_or_else(|| TxLogStorage::state_dir_name(checkpoint_version)); - - let now_ms = current_timestamp_ms(); - let retention_ms = config.log_retention_days as i64 * 24 * 60 * 60 * 1000; - let cp_retention_ms = config.checkpoint_retention_hours as i64 * 60 * 60 * 1000; - - // 2. Delete old version files at or before checkpoint version - let versions = storage.list_versions().await?; - for version in &versions { - if *version > checkpoint_version { - // Post-checkpoint version, keep it - continue; - } - - // Check if the version file is old enough to delete. - // We approximate the version file age by reading its content and checking - // the modification_time of the first AddAction (if any), or we use a - // simpler heuristic: treat all pre-checkpoint versions as eligible if - // the checkpoint itself was created longer ago than the retention period. - // - // For simplicity and correctness, we check the checkpoint's created_time. - // If checkpoint is recent but retention is long, we won't delete. - let checkpoint_age_ms = if last_cp.created_time > 0 { now_ms - last_cp.created_time } else { 0 }; - if checkpoint_age_ms >= retention_ms { - let path = TxLogStorage::version_path(*version); - match storage.delete(&path).await { - Ok(()) => result.deleted_version_files += 1, - Err(_) => {} // Best effort - } - } - } - - // 3. Delete orphaned state directories - let all_entries = storage.list("").await?; - let state_dirs: Vec = all_entries.iter() - .filter_map(|entry| { - // Extract state dir name from paths like "state-v42/_manifest" or "state-v42/manifest-0000.avro" - let entry = entry.trim_start_matches('/'); - if entry.starts_with("state-v") { - entry.split('/').next().map(|s| s.to_string()) - } else { - None - } - }) - .collect::>() - .into_iter() - .collect(); - - for state_dir in &state_dirs { - if *state_dir == checkpoint_state_dir { - // This is the current checkpoint's state dir, keep it - continue; - } - - // Try to read the _manifest.avro to check created_time - let manifest_path = format!("{}/{}", state_dir, super::actions::STATE_MANIFEST_FILENAME); - let should_delete = match storage.get(&manifest_path).await { - Ok(data) => { - match super::avro::state_reader::parse_state_manifest(&data) { - Ok(manifest) => { - let age_ms = now_ms - manifest.created_time; - age_ms >= cp_retention_ms - } - Err(_) => true, // Corrupted manifest, clean it up - } - } - Err(_) => true, // Can't read manifest, consider orphaned - }; - - if should_delete { - // Delete all files in the state directory - let dir_entries = storage.list(state_dir).await.unwrap_or_default(); - for entry in &dir_entries { - let full_path = format!("{}/{}", state_dir, entry); - let _ = storage.delete(&full_path).await; - } - result.deleted_state_dirs += 1; - } - } - - Ok(result) -} - -fn current_timestamp_ms() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as i64) - .unwrap_or(0) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::delta_reader::engine::DeltaStorageConfig; - use crate::txlog::actions::*; - use crate::txlog::version_file; - use std::collections::HashMap; - - fn test_storage(dir: &std::path::Path) -> TxLogStorage { - let config = DeltaStorageConfig::default(); - let url = format!("file://{}", dir.display()); - TxLogStorage::new(&url, &config).unwrap() - } - - fn ensure_txlog_dir(dir: &std::path::Path) { - std::fs::create_dir_all(dir.join("_transaction_log")).unwrap(); - } - - fn make_add_action(path: &str) -> AddAction { - AddAction { - path: path.to_string(), - partition_values: HashMap::new(), - size: 100, - modification_time: 1700000000000, - data_change: true, - stats: None, min_values: None, max_values: None, num_records: Some(10), - footer_start_offset: None, footer_end_offset: None, has_footer_offsets: None, delete_opstamp: None, - split_tags: None, num_merge_ops: None, - doc_mapping_json: None, doc_mapping_ref: None, - uncompressed_size_bytes: None, - time_range_start: None, time_range_end: None, - companion_source_files: None, companion_delta_version: None, - companion_fast_field_mode: None, - } - } - - fn make_file_entry(path: &str, version: i64) -> FileEntry { - FileEntry { - add: make_add_action(path), - added_at_version: version, - added_at_timestamp: 1700000000000, - } - } - - #[tokio::test] - async fn test_gc_preserves_current_and_post_checkpoint() { - let tmp = tempfile::TempDir::new().unwrap(); - ensure_txlog_dir(tmp.path()); - let storage = test_storage(tmp.path()); - - // Write versions 0, 1, 2 - let v0 = vec![ - Action::Protocol(ProtocolAction::v4()), - Action::MetaData(MetadataAction { name: None, description: None, - id: "gc-test".into(), - schema_string: "{}".into(), - partition_columns: vec![], - format: FormatSpec::default(), - configuration: HashMap::new(), - created_time: Some(1700000000000), - }), - Action::Add(make_add_action("s1.split")), - ]; - assert!(version_file::write_version(&storage, 0, &v0).await.unwrap()); - assert!(version_file::write_version(&storage, 1, &[Action::Add(make_add_action("s2.split"))]).await.unwrap()); - assert!(version_file::write_version(&storage, 2, &[Action::Add(make_add_action("s3.split"))]).await.unwrap()); - - // Create checkpoint at version 1 - let entries = vec![ - make_file_entry("s1.split", 0), - make_file_entry("s2.split", 1), - ]; - let metadata = MetadataAction { name: None, description: None, - id: "gc-test".into(), - schema_string: "{}".into(), - partition_columns: vec![], - format: FormatSpec::default(), - configuration: HashMap::new(), - created_time: Some(1700000000000), - }; - crate::txlog::avro::state_writer::write_state_checkpoint( - &storage, 1, &entries, &ProtocolAction::v4(), &metadata, - ).await.unwrap(); - - // Run GC with zero retention (delete immediately) - let gc_config = GcConfig { - log_retention_days: 0, - checkpoint_retention_hours: 0, - }; - let result = garbage_collect(&storage, &gc_config).await.unwrap(); - - // Versions 0 and 1 should be deleted (at or before checkpoint version 1) - // Version 2 should be preserved (post-checkpoint) - assert!(result.deleted_version_files >= 1, "Should delete at least 1 old version file"); - - // Version 2 (post-checkpoint) should still be readable - let v2_read = version_file::read_version(&storage, 2).await; - assert!(v2_read.is_ok(), "Post-checkpoint version 2 should be preserved"); - } - - #[test] - fn test_gc_config_defaults() { - let config = GcConfig::default(); - assert_eq!(config.log_retention_days, 30); - assert_eq!(config.checkpoint_retention_hours, 2); - } -} diff --git a/native/src/txlog/integration_tests.rs b/native/src/txlog/integration_tests.rs index 940dd3d2..8f51cab3 100644 --- a/native/src/txlog/integration_tests.rs +++ b/native/src/txlog/integration_tests.rs @@ -821,6 +821,7 @@ fn test_serialize_snapshot_info() { metadata: make_metadata("snapshot-test"), state_dir: "state-v42".to_string(), tombstones: vec![], + post_checkpoint_actions: None, }; let buf = serialization::serialize_snapshot_info(&info); diff --git a/native/src/txlog/jni.rs b/native/src/txlog/jni.rs index 777fb234..223ac497 100644 --- a/native/src/txlog/jni.rs +++ b/native/src/txlog/jni.rs @@ -8,48 +8,22 @@ use jni::objects::{JClass, JObject, JString}; use jni::sys::{jbyteArray, jint, jlong}; use jni::JNIEnv; -use crate::common::{buffer_to_jbytearray, build_storage_config, extract_optional_jstring, extract_string, to_java_exception}; +use crate::common::{buffer_to_jbytearray, build_storage_config, extract_optional_jstring, to_java_exception}; use crate::runtime_manager::block_on_operation; use super::distributed; use super::serialization; -/// Extract cache, checkpoint, and concurrency config from the Java Map into a Rust HashMap. +/// Extract the full config from the Java Map into a Rust HashMap. /// -/// Recognised keys (all optional): -/// - `cache.ttl.ms` / `cache_ttl_ms` — all-tier TTL override in milliseconds -/// - `cache.version.ttl.ms` — version cache TTL in ms -/// - `cache.snapshot.ttl.ms` — snapshot cache TTL in ms -/// - `cache.file_list.ttl.ms` — file-list cache TTL in ms -/// - `cache.metadata.ttl.ms` — metadata cache TTL in ms -/// - `cache.version.capacity` — version cache max entries -/// - `cache.snapshot.capacity` — snapshot cache max entries -/// - `cache.file_list.capacity` — file-list cache max entries -/// - `cache.enabled` — "false" to disable caching -/// - `max_concurrent_reads` — max parallel object-store GETs (default 32) -/// - `checkpoint_interval` / `checkpoint.interval` — checkpoint frequency +/// Copies the entire map rather than whitelisting specific keys. A whitelist +/// silently drops any key it doesn't enumerate — that already caused +/// `session.timezone.offset.seconds` to be dropped, disabling timestamp-timezone +/// data skipping. Copying everything is simpler and future-proof: new config keys +/// (cache TTLs/capacities, `max_concurrent_reads`, `checkpoint_interval`, +/// `session.timezone.offset.seconds`, etc.) are passed through automatically. fn extract_extended_config(env: &mut JNIEnv, config_map: &JObject) -> std::collections::HashMap { - let mut map = std::collections::HashMap::new(); - for key in &[ - "cache.ttl.ms", - "cache_ttl_ms", - "cache.version.ttl.ms", - "cache.snapshot.ttl.ms", - "cache.file_list.ttl.ms", - "cache.metadata.ttl.ms", - "cache.version.capacity", - "cache.snapshot.capacity", - "cache.file_list.capacity", - "cache.enabled", - "max_concurrent_reads", - "checkpoint_interval", - "checkpoint.interval", - ] { - if let Some(val) = extract_string(env, config_map, key) { - map.insert(key.to_string(), val); - } - } - map + crate::common::extract_hashmap(env, config_map).unwrap_or_default() } // ============================================================================ diff --git a/native/src/txlog/list_files.rs b/native/src/txlog/list_files.rs index 616e3114..b86d9f33 100644 --- a/native/src/txlog/list_files.rs +++ b/native/src/txlog/list_files.rs @@ -107,9 +107,12 @@ pub async unsafe fn list_files_arrow_ffi( // Extract field types from the table schema for type-aware data skipping. // The Rust layer owns the schema — field types are ALWAYS derived from // MetadataAction.schema_string, never passed from the JVM. + // Clamp to a char boundary — a byte-index slice would panic if byte 200 lands + // in the middle of a multi-byte UTF-8 sequence. + let schema_preview: String = snapshot.metadata.schema_string.chars().take(200).collect(); crate::debug_println!("LIST_FILES: schema_string length={}, first 200 chars: {}", snapshot.metadata.schema_string.len(), - &snapshot.metadata.schema_string[..std::cmp::min(200, snapshot.metadata.schema_string.len())]); + schema_preview); let field_types = extract_field_types_from_schema(&snapshot.metadata.schema_string); crate::debug_println!("LIST_FILES: extracted field_types: {:?}", field_types); @@ -167,12 +170,26 @@ pub async unsafe fn list_files_arrow_ffi( snapshot.tombstones.len(), before, checkpoint_entries.len()); } - // 4. Read post-checkpoint changes and merge - let changes = distributed::read_post_checkpoint_changes( - table_path, config, &snapshot.post_checkpoint_version_paths, &metadata_config, - ).await?; + // 4. Compute post-checkpoint changes and merge. + // Reuse the version actions the snapshot already parsed when available (E1): + // the snapshot-info path reads these same files for protocol/metadata override, + // so re-fetching them here would be a 2× GET per query. Fall back to a fresh + // read only when the snapshot didn't capture them. + let changes = match &snapshot.post_checkpoint_actions { + Some(actions) => distributed::changes_from_versioned_actions(actions, &metadata_config), + None => distributed::read_post_checkpoint_changes( + table_path, config, &snapshot.post_checkpoint_version_paths, &metadata_config, + ).await?, + }; - // Apply adds/removes to checkpoint state + // Apply adds/removes to checkpoint state. + // + // INVARIANT: a removed path is never re-added. Splits are immutable and new data + // always gets a fresh path, so applying all adds and then all removes (without + // interleaving them by version) is correct — no add here can resurrect a path + // that a later remove intends to delete. If that invariant were ever violated, + // an add followed by a same-batch remove of the same path would incorrectly drop + // a live file (or vice versa). See TXLOG_MODULE_REVIEW (C1). let mut file_map: HashMap = HashMap::new(); for entry in checkpoint_entries { file_map.insert(entry.add.path.clone(), entry); diff --git a/native/src/txlog/metrics.rs b/native/src/txlog/metrics.rs deleted file mode 100644 index 2089e0cd..00000000 --- a/native/src/txlog/metrics.rs +++ /dev/null @@ -1,69 +0,0 @@ -// txlog/metrics.rs - Write retries, cache stats, timing - -use std::sync::atomic::{AtomicU64, Ordering}; - -#[derive(Debug, Default)] -pub struct TxLogMetrics { - // Write metrics - pub write_attempts: AtomicU64, - pub write_conflicts: AtomicU64, - pub write_successes: AtomicU64, - pub total_retry_delay_ms: AtomicU64, - - // Read metrics - pub versions_read: AtomicU64, - pub manifests_read: AtomicU64, - pub file_entries_loaded: AtomicU64, - - // Timing - pub last_list_files_ms: AtomicU64, - pub last_write_ms: AtomicU64, - pub last_checkpoint_ms: AtomicU64, -} - -#[derive(Debug, Clone, serde::Serialize)] -pub struct TxLogMetricsSnapshot { - pub write_attempts: u64, - pub write_conflicts: u64, - pub write_successes: u64, - pub avg_retry_delay_ms: f64, - pub versions_read: u64, - pub manifests_read: u64, - pub file_entries_loaded: u64, - pub last_list_files_ms: u64, - pub last_write_ms: u64, - pub last_checkpoint_ms: u64, -} - -impl TxLogMetrics { - pub fn snapshot(&self) -> TxLogMetricsSnapshot { - let attempts = self.write_attempts.load(Ordering::Relaxed); - let total_delay = self.total_retry_delay_ms.load(Ordering::Relaxed); - let conflicts = self.write_conflicts.load(Ordering::Relaxed); - TxLogMetricsSnapshot { - write_attempts: attempts, - write_conflicts: conflicts, - write_successes: self.write_successes.load(Ordering::Relaxed), - avg_retry_delay_ms: if conflicts > 0 { total_delay as f64 / conflicts as f64 } else { 0.0 }, - versions_read: self.versions_read.load(Ordering::Relaxed), - manifests_read: self.manifests_read.load(Ordering::Relaxed), - file_entries_loaded: self.file_entries_loaded.load(Ordering::Relaxed), - last_list_files_ms: self.last_list_files_ms.load(Ordering::Relaxed), - last_write_ms: self.last_write_ms.load(Ordering::Relaxed), - last_checkpoint_ms: self.last_checkpoint_ms.load(Ordering::Relaxed), - } - } - - pub fn reset(&self) { - self.write_attempts.store(0, Ordering::Relaxed); - self.write_conflicts.store(0, Ordering::Relaxed); - self.write_successes.store(0, Ordering::Relaxed); - self.total_retry_delay_ms.store(0, Ordering::Relaxed); - self.versions_read.store(0, Ordering::Relaxed); - self.manifests_read.store(0, Ordering::Relaxed); - self.file_entries_loaded.store(0, Ordering::Relaxed); - self.last_list_files_ms.store(0, Ordering::Relaxed); - self.last_write_ms.store(0, Ordering::Relaxed); - self.last_checkpoint_ms.store(0, Ordering::Relaxed); - } -} diff --git a/native/src/txlog/mod.rs b/native/src/txlog/mod.rs index 9c52474e..1369b1e1 100644 --- a/native/src/txlog/mod.rs +++ b/native/src/txlog/mod.rs @@ -14,11 +14,9 @@ pub mod cache; pub mod compression; pub mod distributed; pub mod error; -pub mod garbage_collection; pub mod jni; pub mod list_files; pub mod log_replay; -pub mod metrics; pub mod partition_pruning; pub mod purge; pub mod schema_dedup; diff --git a/native/src/txlog/parallel_bench_tests.rs b/native/src/txlog/parallel_bench_tests.rs index 9cf2de10..dbf1d261 100644 --- a/native/src/txlog/parallel_bench_tests.rs +++ b/native/src/txlog/parallel_bench_tests.rs @@ -464,9 +464,14 @@ async fn test_probe_versions_since_finds_contiguous() { assert_eq!(versions, vec![3, 4, 5], "should find versions 3, 4, 5"); } -/// Verify probe_versions_since stops at a gap (version 4 missing). +/// Verify probe_versions_since recovers versions past a gap (version 4 missing). +/// +/// A mid-sequence gap indicates an operator-deleted version file. Naively stopping +/// at the gap would hide every later commit until TTL expiry; instead the probe +/// falls back to a full LIST when it sees an existing version beyond the gap, so +/// version 5 is still recovered. #[tokio::test] -async fn test_probe_versions_since_stops_at_gap() { +async fn test_probe_versions_since_recovers_past_gap() { use crate::txlog::storage::TxLogStorage; use crate::delta_reader::engine::DeltaStorageConfig; @@ -485,5 +490,5 @@ async fn test_probe_versions_since_stops_at_gap() { let storage = TxLogStorage::new(&table_path, &config).unwrap(); let versions = super::distributed::probe_versions_since(&storage, 2).await.unwrap(); - assert_eq!(versions, vec![3], "should stop at gap — only version 3"); + assert_eq!(versions, vec![3, 5], "should recover version 5 past the gap at 4"); } diff --git a/native/src/txlog/partition_pruning.rs b/native/src/txlog/partition_pruning.rs index 6f62fb52..33ed24f7 100644 --- a/native/src/txlog/partition_pruning.rs +++ b/native/src/txlog/partition_pruning.rs @@ -60,12 +60,18 @@ impl PartitionFilter { }) } PartitionFilter::Neq { column, value } => { + // NOTE: diverges from SQL NULL semantics. A missing partition value + // returns `true` here (SQL would evaluate `NULL != x` as NULL, i.e. + // filtered out). This errs toward INCLUDING files, so results remain + // correct as long as the query engine re-filters rows after listing. partition_values.get(column).map_or(true, |v| v != value) } PartitionFilter::In { column, values } => { partition_values.get(column).map_or(false, |v| values.contains(v)) } PartitionFilter::IsNull { column } => { + // Proxies "null" as the empty string — a missing value or an empty + // value both count as null. Errs toward including files. partition_values.get(column).map_or(true, |v| v.is_empty()) } PartitionFilter::IsNotNull { column } => { @@ -209,13 +215,20 @@ impl PartitionFilter { ) -> bool { match self { PartitionFilter::Eq { column, value } => { - // Skip if value < min OR value > max + // Skip if value < min OR value > max. + // Min side needs no truncation guard: string stats are truncated to a + // prefix, so the actual min is >= the stored min; value < stored min ⇒ + // value < actual min ⇒ safe to skip. The max side DOES need a guard, + // since the actual max may extend past the stored (truncated) max. if let Some(min_val) = min_values.get(column) { if compare_values(value, min_val) == std::cmp::Ordering::Less { return true; } } if let Some(max_val) = max_values.get(column) { + if value.len() > max_val.len() && value.starts_with(max_val.as_str()) { + return false; // max may be truncated; actual max could be >= value + } if compare_values(value, max_val) == std::cmp::Ordering::Greater { return true; } @@ -232,8 +245,13 @@ impl PartitionFilter { } } PartitionFilter::Gt { column, value } => { - // Skip if max <= value (no value in file is > value) + // Skip if max <= value (no value in file is > value). + // Truncation guard: the stored max may be a prefix of a larger actual + // max, so don't skip when value is a proper prefix-extension of max. if let Some(max_val) = max_values.get(column) { + if value.len() > max_val.len() && value.starts_with(max_val.as_str()) { + return false; // max may be truncated + } return compare_values(max_val, value) != std::cmp::Ordering::Greater; } false @@ -336,53 +354,78 @@ impl PartitionFilter { ) -> bool { match self { PartitionFilter::Eq { column, value } => { + use std::cmp::Ordering; let ftype = ft.get(column).map(|s| s.as_str()); if let Some(min_val) = min_values.get(column) { - if compare_values_typed(value, min_val, ftype, tz_offset_secs) == std::cmp::Ordering::Less { + // Skip if value < min. Unknown (None) → don't skip. + if compare_values_typed(value, min_val, ftype, tz_offset_secs) == Some(Ordering::Less) { return true; } } if let Some(max_val) = max_values.get(column) { - if compare_values_typed(value, max_val, ftype, tz_offset_secs) == std::cmp::Ordering::Greater { + // Truncation guard on the max side (string stats only). + if value.len() > max_val.len() && value.starts_with(max_val.as_str()) { + return false; // max may be truncated + } + // Skip if value > max. Unknown (None) → don't skip. + if compare_values_typed(value, max_val, ftype, tz_offset_secs) == Some(Ordering::Greater) { return true; } } false } PartitionFilter::Gt { column, value } => { + use std::cmp::Ordering; let ftype = ft.get(column).map(|s| s.as_str()); if let Some(max_val) = max_values.get(column) { - return compare_values_typed(max_val, value, ftype, tz_offset_secs) != std::cmp::Ordering::Greater; + // Truncation guard: stored max may be a prefix of a larger actual max. + if value.len() > max_val.len() && value.starts_with(max_val.as_str()) { + return false; // max may be truncated + } + // Skip if max <= value (no value in file is > value). Unknown → don't skip. + return matches!( + compare_values_typed(max_val, value, ftype, tz_offset_secs), + Some(Ordering::Less) | Some(Ordering::Equal) + ); } false } PartitionFilter::Gte { column, value } => { + use std::cmp::Ordering; let ftype = ft.get(column).map(|s| s.as_str()); if let Some(max_val) = max_values.get(column) { if value.len() > max_val.len() && value.starts_with(max_val.as_str()) { return false; } - return compare_values_typed(max_val, value, ftype, tz_offset_secs) == std::cmp::Ordering::Less; + // Skip if max < value. Unknown → don't skip. + return compare_values_typed(max_val, value, ftype, tz_offset_secs) == Some(Ordering::Less); } false } PartitionFilter::Lt { column, value } => { + use std::cmp::Ordering; let ftype = ft.get(column).map(|s| s.as_str()); if let Some(min_val) = min_values.get(column) { if value.len() > min_val.len() && value.starts_with(min_val.as_str()) { return false; } - return compare_values_typed(min_val, value, ftype, tz_offset_secs) != std::cmp::Ordering::Less; + // Skip if min >= value (no value in file is < value). Unknown → don't skip. + return matches!( + compare_values_typed(min_val, value, ftype, tz_offset_secs), + Some(Ordering::Greater) | Some(Ordering::Equal) + ); } false } PartitionFilter::Lte { column, value } => { + use std::cmp::Ordering; let ftype = ft.get(column).map(|s| s.as_str()); if let Some(min_val) = min_values.get(column) { if value.len() > min_val.len() && value.starts_with(min_val.as_str()) { return false; } - return compare_values_typed(min_val, value, ftype, tz_offset_secs) == std::cmp::Ordering::Greater; + // Skip if min > value. Unknown → don't skip. + return compare_values_typed(min_val, value, ftype, tz_offset_secs) == Some(Ordering::Greater); } false } @@ -427,13 +470,17 @@ pub fn prune_manifests<'a>( /// - "date": converts date strings ("2024-01-15") to epoch days for comparison with /// stat values stored as epoch day integers or epoch microseconds /// - "timestamp": converts timestamp strings to epoch microseconds using the session -/// timezone offset. If no timezone offset is provided, falls back to conservative -/// comparison (Ordering::Equal = never skip). +/// timezone offset. If no timezone offset is provided, the comparison is *unknown* +/// and returns `None` so callers can fail safe (never skip). /// - For all other types (or when field_type is None): falls through to `compare_values` /// +/// Returns `None` when the comparison cannot be determined (e.g. a bare datetime +/// filter value with no session timezone available). A sentinel value in a total +/// order cannot express uncertainty, so callers MUST treat `None` as "don't skip". +/// /// `tz_offset_seconds`: session timezone as seconds east of UTC (e.g., -18000 for EST). /// Passed from JVM via config map key "session.timezone.offset.seconds". -pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offset_secs: Option) -> std::cmp::Ordering { +pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offset_secs: Option) -> Option { match field_type { Some("date") => { // Try to normalize both values to epoch days for comparison. @@ -441,10 +488,10 @@ pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offse let a_days = parse_as_epoch_days(a); let b_days = parse_as_epoch_days(b); if let (Some(ad), Some(bd)) = (a_days, b_days) { - return ad.cmp(&bd); + return Some(ad.cmp(&bd)); } // Fallback to default comparison if either fails to parse - compare_values(a, b) + Some(compare_values(a, b)) } Some("timestamp") | Some("timestamp_ntz") => { // Stats are stored as epoch micros. Filter values may be bare datetime @@ -453,7 +500,7 @@ pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offse // // When both sides are numeric, compare directly as epoch micros. // When a filter is a datetime string, use tz_offset_secs to convert. - // Without a timezone offset, be conservative (don't skip). + // Without a timezone offset, the answer is unknown (return None). let a_is_numeric = a.parse::().is_ok(); let b_is_numeric = b.parse::().is_ok(); @@ -461,7 +508,7 @@ pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offse let a_micros = parse_as_epoch_micros(a); let b_micros = parse_as_epoch_micros(b); if let (Some(am), Some(bm)) = (a_micros, b_micros) { - return am.cmp(&bm); + return Some(am.cmp(&bm)); } } @@ -475,7 +522,7 @@ pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offse let a_micros = parse_as_epoch_micros(a); let b_micros = parse_as_epoch_micros(b); if let (Some(am), Some(bm)) = (a_micros, b_micros) { - return am.cmp(&bm); + return Some(am.cmp(&bm)); } } @@ -484,14 +531,14 @@ pub fn compare_values_typed(a: &str, b: &str, field_type: Option<&str>, tz_offse let a_micros = parse_as_epoch_micros_with_tz(a, tz_offset); let b_micros = parse_as_epoch_micros_with_tz(b, tz_offset); if let (Some(am), Some(bm)) = (a_micros, b_micros) { - return am.cmp(&bm); + return Some(am.cmp(&bm)); } } - // No timezone available — conservative (never skip) - std::cmp::Ordering::Equal + // No timezone available — comparison is unknown; caller must not skip. + None } - _ => compare_values(a, b), + _ => Some(compare_values(a, b)), } } @@ -547,7 +594,12 @@ fn parse_as_epoch_days(s: &str) -> Option { /// This matches Spark's convention: stats are stored as epoch microseconds, /// filter values may be epoch seconds or millis depending on the Spark version. fn parse_as_epoch_micros(s: &str) -> Option { - // Try integer first — determine unit by magnitude + // Try integer first — determine unit by magnitude. + // NOTE: the magnitude heuristic only works for positive epoch values. Spark + // stores timestamp stats as epoch MICROSECONDS, so a negative value is a + // pre-1970 timestamp already in micros — pass it through rather than collapsing + // all pre-epoch timestamps to 0 (which would make them compare equal and + // produce wrong skips for historical data). if let Ok(v) = s.parse::() { if v > 1_000_000_000_000_000 { return Some(v); // Microseconds @@ -555,27 +607,40 @@ fn parse_as_epoch_micros(s: &str) -> Option { return Some(v * 1_000); // Milliseconds → microseconds } else if v > 0 { return Some(v * 1_000_000); // Seconds → microseconds + } else if v == 0 { + return Some(0); // Exactly the epoch } - return Some(0); // Zero or negative — epoch start + return Some(v); // Negative → pre-1970 epoch micros (Spark's storage unit) } - // Try full ISO-8601 with timezone (RFC 3339): - // "2023-11-07T05:00:00Z" - // "2023-11-07T05:00:00+05:00" - // "2023-11-07T05:00:00.123456Z" + // Non-integer: delegate to the shared datetime-string parser. + parse_datetime_string_to_micros(s) +} + +/// Parse a non-integer datetime string to epoch microseconds (UTC). +/// +/// Handles, in order: +/// - RFC 3339 / ISO-8601 with an explicit zone ("2023-11-07T05:00:00Z", "…+05:00") +/// - Bare datetime, no zone ("2023-11-07 05:00:00", "2023-11-07T05:00:00", with +/// optional fractional seconds) — treated as UTC +/// - Date-only ("2023-11-07") — midnight UTC +/// +/// Zoneless values are treated as UTC (no session timezone is available at the +/// stats/export layer). Shared by `parse_as_epoch_micros` here and +/// `arrow_ffi::time_range_to_epoch_millis` so the accepted-format list can't drift +/// between the two (TXLOG_MODULE_REVIEW follow-up: de-dup + date-only coverage). +pub(crate) fn parse_datetime_string_to_micros(s: &str) -> Option { if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) { return Some(dt.timestamp_micros()); } - - // Try JDBC / Spark format (no timezone — treated as UTC): - // "2023-11-07 05:00:00" - // "2023-11-07T05:00:00" let normalized = s.replace('T', " "); - if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(&normalized, "%Y-%m-%d %H:%M:%S") { + if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(&normalized, "%Y-%m-%d %H:%M:%S") + .or_else(|_| chrono::NaiveDateTime::parse_from_str(&normalized, "%Y-%m-%d %H:%M:%S%.f")) + { return Some(dt.and_utc().timestamp_micros()); } - // With fractional seconds - if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(&normalized, "%Y-%m-%d %H:%M:%S%.f") { - return Some(dt.and_utc().timestamp_micros()); + // Date-only → midnight UTC. + if let Ok(d) = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") { + return Some(d.and_hms_opt(0, 0, 0)?.and_utc().timestamp_micros()); } None } @@ -1326,6 +1391,54 @@ mod tests { assert!(!f.can_skip_by_stats_typed(&min, &max, &ft, None)); } + // ------------------------------------------------------------------------ + // C4 regression: an *unknown* typed comparison (bare-datetime filter with no + // session timezone) must never cause a file to be skipped. Before the + // Option fix, `compare_values_typed` returned Ordering::Equal as a + // sentinel and Gt/Lt inverted it into "skip everything". + // ------------------------------------------------------------------------ + + #[test] + fn test_timestamp_gt_bare_datetime_no_tz_does_not_skip() { + // Stats stored as epoch micros; filter is a bare datetime string with no + // session timezone available. The comparison is unknown → must NOT skip. + let (min, max) = stats(&[("ts", "1699315200000000")], &[("ts", "1699401600000000")]); + let ft: HashMap = [("ts".into(), "timestamp".into())].into(); + let f = PartitionFilter::Gt { column: "ts".into(), value: "2023-11-07 05:00:00".into() }; + assert!( + !f.can_skip_by_stats_typed(&min, &max, &ft, None), + "Gt with unknown comparison must not skip live files" + ); + } + + #[test] + fn test_timestamp_lt_bare_datetime_no_tz_does_not_skip() { + let (min, max) = stats(&[("ts", "1699315200000000")], &[("ts", "1699401600000000")]); + let ft: HashMap = [("ts".into(), "timestamp".into())].into(); + let f = PartitionFilter::Lt { column: "ts".into(), value: "2023-11-07 05:00:00".into() }; + assert!( + !f.can_skip_by_stats_typed(&min, &max, &ft, None), + "Lt with unknown comparison must not skip live files" + ); + } + + #[test] + fn test_timestamp_gt_bare_datetime_with_tz_skips_correctly() { + // Once a session timezone is supplied, the bare-datetime filter can be + // resolved and data skipping works. Stats span 2023-11-07..2023-11-08 UTC; + // filter ts > 2023-11-09 00:00:00 UTC (tz_offset 0) → max < filter → skip. + let (min, max) = stats(&[("ts", "1699315200000000")], &[("ts", "1699401600000000")]); + let ft: HashMap = [("ts".into(), "timestamp".into())].into(); + let f = PartitionFilter::Gt { column: "ts".into(), value: "2023-11-09 00:00:00".into() }; + assert!( + f.can_skip_by_stats_typed(&min, &max, &ft, Some(0)), + "with a session tz, a filter beyond max should skip" + ); + // And a filter below max must not skip. + let f2 = PartitionFilter::Gt { column: "ts".into(), value: "2023-11-06 00:00:00".into() }; + assert!(!f2.can_skip_by_stats_typed(&min, &max, &ft, Some(0))); + } + // ======================================================================== // Deserialization tests // ======================================================================== diff --git a/native/src/txlog/purge.rs b/native/src/txlog/purge.rs index f36ab8d6..9cd8909a 100644 --- a/native/src/txlog/purge.rs +++ b/native/src/txlog/purge.rs @@ -3,7 +3,7 @@ // Provides retention-based version/state expiration and retained file enumeration // for the Spark-side anti-join purge algorithm. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use parking_lot::Mutex; @@ -80,9 +80,14 @@ pub async fn list_retained_versions( let version_ts = actions.iter().find_map(|a| match a { Action::Add(add) => Some(add.modification_time), _ => None, - }).unwrap_or(0); - if now_ms - version_ts < retention_ms { - retained.push(*v); + }); + match version_ts { + // Version carries an Add timestamp → use it to judge age. + Some(ts) if now_ms - ts < retention_ms => retained.push(*v), + Some(_) => { /* older than retention → drop from cursor */ } + // No Add (only removes/metadata) → age is unknown; keep (safe + // default, consistent with the parse/read-failure branches). + None => retained.push(*v), } } else { // Can't parse → keep (safe default) @@ -144,7 +149,13 @@ pub async fn open_retained_files_cursor( } } - // Apply post-checkpoint changes from retained versions + // Apply post-checkpoint changes from retained versions. + // Accumulate removes into a set and apply them in a single pass at the + // end rather than an O(entries) `retain` per Remove action — that was + // O(entries × removes) on large tables (TXLOG_MODULE_REVIEW E4). Applying + // all removes together is correct under the never-re-added invariant (a + // removed path is never re-added, so remove-after-add ordering is moot). + let mut removed_paths: HashSet = HashSet::new(); for v in &retained_versions { if *v > cp.version { if let Ok(actions) = version_file::read_version(&storage, *v).await { @@ -159,7 +170,7 @@ pub async fn open_retained_files_cursor( }); } Action::Remove(r) => { - all_entries.retain(|e| e.add.path != r.path); + removed_paths.insert(r.path); } _ => {} } @@ -167,6 +178,9 @@ pub async fn open_retained_files_cursor( } } } + if !removed_paths.is_empty() { + all_entries.retain(|e| !removed_paths.contains(&e.add.path)); + } } } else { // No checkpoint — replay all retained versions @@ -180,7 +194,11 @@ pub async fn open_retained_files_cursor( all_entries = replay_result.files; } - // Deduplicate by path (keep latest version) + // Deduplicate by path. `retain(|e| seen.insert(...))` keeps the FIRST occurrence + // of each path (entries are accumulated in ascending version order). This relies + // on the module invariant that a path is never re-added once removed — splits are + // immutable and new data always gets a fresh path — so no genuine duplicates exist + // here and first-vs-last is moot. See the invariant note in TXLOG_MODULE_REVIEW (C1). let mut seen = std::collections::HashSet::new(); all_entries.retain(|e| seen.insert(e.add.path.clone())); diff --git a/native/src/txlog/serialization.rs b/native/src/txlog/serialization.rs index 0c48ca07..368ae517 100644 --- a/native/src/txlog/serialization.rs +++ b/native/src/txlog/serialization.rs @@ -265,44 +265,50 @@ pub fn serialize_changes(changes: &TxLogChanges) -> Vec { // Skip actions (expanded fields for GAP-6) for skip in &changes.skip_actions { offsets.push(buf.len() as u32); - let mut fc: u16 = 4; // path, reason, skip_count, skip_timestamp - if !skip.operation.is_empty() { fc += 1; } - if skip.partition_values.is_some() { fc += 1; } - if skip.size.is_some() { fc += 1; } - if skip.retry_after.is_some() { fc += 1; } - buf.extend_from_slice(&fc.to_ne_bytes()); - write_field_header(&mut buf, "path", FIELD_TYPE_TEXT, 1); - write_string(&mut buf, &skip.path); - write_field_header(&mut buf, "skip_timestamp", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&skip.skip_timestamp.to_ne_bytes()); - write_field_header(&mut buf, "reason", FIELD_TYPE_TEXT, 1); - write_string(&mut buf, &skip.reason); - write_field_header(&mut buf, "skip_count", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&(skip.skip_count as i64).to_ne_bytes()); - if !skip.operation.is_empty() { - let op = &skip.operation; - write_field_header(&mut buf, "operation", FIELD_TYPE_TEXT, 1); - write_string(&mut buf, op); - } - if let Some(ref pv) = skip.partition_values { - write_field_header(&mut buf, "partition_values", FIELD_TYPE_JSON, 1); - let json = serde_json::to_string(pv).unwrap_or_else(|_| "{}".to_string()); - write_string(&mut buf, &json); - } - if let Some(v) = skip.size { - write_field_header(&mut buf, "size", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&v.to_ne_bytes()); - } - if let Some(v) = skip.retry_after { - write_field_header(&mut buf, "retry_after", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&v.to_ne_bytes()); - } + serialize_skip_action(&mut buf, skip); } write_footer(&mut buf, &offsets); buf } +/// Serialize a single SkipAction document into `buf` (one TANT document with a +/// variable field count). Shared by `serialize_changes` and `serialize_skip_actions` +/// so the two encoders can never drift (TXLOG_MODULE_REVIEW D6). +fn serialize_skip_action(buf: &mut Vec, skip: &super::actions::SkipAction) { + let mut fc: u16 = 4; // path, skip_timestamp, reason, skip_count + if !skip.operation.is_empty() { fc += 1; } + if skip.partition_values.is_some() { fc += 1; } + if skip.size.is_some() { fc += 1; } + if skip.retry_after.is_some() { fc += 1; } + buf.extend_from_slice(&fc.to_ne_bytes()); + write_field_header(buf, "path", FIELD_TYPE_TEXT, 1); + write_string(buf, &skip.path); + write_field_header(buf, "skip_timestamp", FIELD_TYPE_INTEGER, 1); + buf.extend_from_slice(&skip.skip_timestamp.to_ne_bytes()); + write_field_header(buf, "reason", FIELD_TYPE_TEXT, 1); + write_string(buf, &skip.reason); + write_field_header(buf, "skip_count", FIELD_TYPE_INTEGER, 1); + buf.extend_from_slice(&(skip.skip_count as i64).to_ne_bytes()); + if !skip.operation.is_empty() { + write_field_header(buf, "operation", FIELD_TYPE_TEXT, 1); + write_string(buf, &skip.operation); + } + if let Some(ref pv) = skip.partition_values { + write_field_header(buf, "partition_values", FIELD_TYPE_JSON, 1); + let json = serde_json::to_string(pv).unwrap_or_else(|_| "{}".to_string()); + write_string(buf, &json); + } + if let Some(v) = skip.size { + write_field_header(buf, "size", FIELD_TYPE_INTEGER, 1); + buf.extend_from_slice(&v.to_ne_bytes()); + } + if let Some(v) = skip.retry_after { + write_field_header(buf, "retry_after", FIELD_TYPE_INTEGER, 1); + buf.extend_from_slice(&v.to_ne_bytes()); + } +} + // ============================================================================ // WriteResult serialization // ============================================================================ @@ -372,38 +378,7 @@ pub fn serialize_skip_actions(skips: &[super::actions::SkipAction]) -> Vec { let mut offsets = Vec::with_capacity(skips.len()); for skip in skips { offsets.push(buf.len() as u32); - let mut fc: u16 = 4; // path, skip_timestamp, reason, skip_count - if !skip.operation.is_empty() { fc += 1; } - if skip.partition_values.is_some() { fc += 1; } - if skip.size.is_some() { fc += 1; } - if skip.retry_after.is_some() { fc += 1; } - buf.extend_from_slice(&fc.to_ne_bytes()); - write_field_header(&mut buf, "path", FIELD_TYPE_TEXT, 1); - write_string(&mut buf, &skip.path); - write_field_header(&mut buf, "skip_timestamp", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&skip.skip_timestamp.to_ne_bytes()); - write_field_header(&mut buf, "reason", FIELD_TYPE_TEXT, 1); - write_string(&mut buf, &skip.reason); - write_field_header(&mut buf, "skip_count", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&(skip.skip_count as i64).to_ne_bytes()); - if !skip.operation.is_empty() { - let op = &skip.operation; - write_field_header(&mut buf, "operation", FIELD_TYPE_TEXT, 1); - write_string(&mut buf, op); - } - if let Some(ref pv) = skip.partition_values { - write_field_header(&mut buf, "partition_values", FIELD_TYPE_JSON, 1); - let json = serde_json::to_string(pv).unwrap_or_else(|_| "{}".to_string()); - write_string(&mut buf, &json); - } - if let Some(v) = skip.size { - write_field_header(&mut buf, "size", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&v.to_ne_bytes()); - } - if let Some(v) = skip.retry_after { - write_field_header(&mut buf, "retry_after", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&v.to_ne_bytes()); - } + serialize_skip_action(&mut buf, skip); } write_footer(&mut buf, &offsets); diff --git a/native/src/txlog/storage.rs b/native/src/txlog/storage.rs index 4a2e42a6..79b570a1 100644 --- a/native/src/txlog/storage.rs +++ b/native/src/txlog/storage.rs @@ -50,7 +50,12 @@ impl TxLogStorage { let path = self.full_path(relative_path); debug_println!("📖 TXLOG_STORAGE: get {}", path); let result = self.store.get(&path).await - .map_err(|e| TxLogError::Storage(anyhow::anyhow!("GET {}: {}", path, e)))?; + .map_err(|e| match e { + // Preserve NotFound structurally so callers can match on the variant + // instead of substring-matching the message (TXLOG_MODULE_REVIEW D4). + object_store::Error::NotFound { .. } => TxLogError::NotFound { path: path.to_string() }, + other => TxLogError::Storage(anyhow::anyhow!("GET {}: {}", path, other)), + })?; result.bytes().await .map_err(|e| TxLogError::Storage(anyhow::anyhow!("Read bytes {}: {}", path, e))) } diff --git a/native/src/txlog/tombstone_distributor.rs b/native/src/txlog/tombstone_distributor.rs index d03eb15e..681da48a 100644 --- a/native/src/txlog/tombstone_distributor.rs +++ b/native/src/txlog/tombstone_distributor.rs @@ -1,13 +1,19 @@ -// txlog/tombstone_distributor.rs - Partition-aware tombstone distribution and selective compaction +// txlog/tombstone_distributor.rs - Tombstone filtering and compaction triggering // -// Matches Scala's TombstoneDistributor: distributes tombstones to manifests -// based on partition bounds, enabling selective compaction that only rewrites -// dirty manifests while keeping clean ones intact. +// Matches Scala's TombstoneDistributor: filters tombstoned entries out of a live +// set and decides when a state manifest needs compaction. +// +// NOTE: the partition-aware *selective* compaction helpers +// (`distribute_tombstones_to_manifests`, `selective_partition`, +// `is_selective_compaction_beneficial`) were removed when the auto-checkpoint path +// collapsed selective compaction into full compaction (TXLOG_MODULE_REVIEW E3): +// the selective branch read every manifest and produced a checkpoint identical to +// full compaction, so it bought nothing. Reintroduce them only alongside a real +// selective writer that reuses kept manifest refs. use std::collections::HashSet; -use super::actions::{FileEntry, ManifestInfo, PartitionBounds}; -use super::partition_pruning::compare_values; +use super::actions::FileEntry; /// Default tombstone ratio threshold for triggering compaction. pub const COMPACTION_TOMBSTONE_THRESHOLD: f64 = 0.10; @@ -15,9 +21,6 @@ pub const COMPACTION_TOMBSTONE_THRESHOLD: f64 = 0.10; /// Default max manifest count before fragmentation compaction. pub const COMPACTION_MAX_MANIFESTS: usize = 20; -/// Minimum savings ratio for selective compaction to be beneficial. -const SELECTIVE_COMPACTION_SAVINGS_RATIO: f64 = 0.10; - /// Filter out entries whose paths appear in the removed set. pub fn filter_tombstoned_entries( entries: &[FileEntry], @@ -38,147 +41,6 @@ pub fn distribute_tombstones( filter_tombstoned_entries(entries, removed_paths) } -/// Distribute tombstones to manifests based on partition bounds. -/// -/// For each manifest, counts how many tombstones fall within its partition bounds. -/// Returns manifests with updated `tombstone_count` and `live_entry_count`. -pub fn distribute_tombstones_to_manifests( - manifests: &[ManifestInfo], - tombstones: &HashSet, - partition_columns: &[String], -) -> Vec { - if tombstones.is_empty() { - return manifests.iter().map(|m| { - let mut updated = m.clone(); - updated.tombstone_count = 0; - updated.live_entry_count = m.file_count; - updated - }).collect(); - } - - // Parse partition values from tombstone paths - let tombstone_partitions: Vec<(&str, std::collections::HashMap)> = tombstones.iter() - .map(|path| (path.as_str(), extract_partition_values(path, partition_columns))) - .collect(); - - manifests.iter().map(|manifest| { - let count = tombstone_partitions.iter() - .filter(|(_, pv)| values_within_bounds(pv, &manifest.partition_bounds)) - .count(); - - // Cap at file_count - let capped = std::cmp::min(count as i64, manifest.file_count); - let live = manifest.file_count - capped; - - let mut updated = manifest.clone(); - updated.tombstone_count = capped; - updated.live_entry_count = live; - updated - }).collect() -} - -/// Check if a set of partition values falls within a manifest's partition bounds. -fn values_within_bounds( - partition_values: &std::collections::HashMap, - bounds: &Option, -) -> bool { - match bounds { - None => partition_values.is_empty(), - Some(bounds) if bounds.min_values.is_empty() && bounds.max_values.is_empty() => { - partition_values.is_empty() - } - Some(bounds) => { - // Check each column in bounds - bounds.min_values.keys() - .chain(bounds.max_values.keys()) - .collect::>() - .iter() - .all(|col| { - match partition_values.get(*col) { - None => true, // Conservative: could match - Some(value) => { - let within_min = bounds.min_values.get(*col) - .map_or(true, |min| compare_values(value, min) != std::cmp::Ordering::Less); - let within_max = bounds.max_values.get(*col) - .map_or(true, |max| compare_values(value, max) != std::cmp::Ordering::Greater); - within_min && within_max - } - } - }) - } - } -} - -/// Extract partition values from a file path. -/// -/// Parses paths like `date=2024-01-01/region=us-east/file.split` -/// into `{"date": "2024-01-01", "region": "us-east"}`. -fn extract_partition_values( - path: &str, - _partition_columns: &[String], -) -> std::collections::HashMap { - let mut values = std::collections::HashMap::new(); - for segment in path.split('/') { - if let Some(eq_pos) = segment.find('=') { - let key = &segment[..eq_pos]; - let val = &segment[eq_pos + 1..]; - if !key.is_empty() && !val.is_empty() { - values.insert(key.to_string(), val.to_string()); - } - } - } - values -} - -/// Partition manifests into keep (clean) and rewrite (dirty) based on tombstone ratio. -/// -/// Returns `(keep, rewrite)` where: -/// - `keep`: manifests with tombstone ratio <= threshold (reused as-is) -/// - `rewrite`: manifests with tombstone ratio > threshold (need rewriting) -pub fn selective_partition( - manifests: &[ManifestInfo], - threshold: f64, -) -> (Vec, Vec) { - let mut keep = Vec::new(); - let mut rewrite = Vec::new(); - - for m in manifests { - if m.file_count == 0 { - keep.push(m.clone()); // Empty manifests — nothing to compact - } else { - let ratio = m.tombstone_count as f64 / m.file_count as f64; - if ratio <= threshold { - keep.push(m.clone()); - } else { - rewrite.push(m.clone()); - } - } - } - - (keep, rewrite) -} - -/// Check if selective compaction is beneficial (saves at least 10% of entries). -pub fn is_selective_compaction_beneficial( - keep: &[ManifestInfo], - rewrite: &[ManifestInfo], -) -> bool { - if keep.is_empty() || rewrite.is_empty() { - return false; - } - - let entries_to_keep: i64 = keep.iter().map(|m| m.file_count).sum(); - let entries_to_rewrite: i64 = rewrite.iter().map(|m| m.file_count).sum(); - let total = entries_to_keep + entries_to_rewrite; - - if total == 0 { - return false; - } - - let savings_ratio = entries_to_keep as f64 / total as f64; - savings_ratio >= SELECTIVE_COMPACTION_SAVINGS_RATIO -} - /// Check if a state manifest needs compaction. /// /// Compaction is triggered when: @@ -203,7 +65,7 @@ pub fn needs_compaction( #[cfg(test)] mod tests { use super::*; - use crate::txlog::actions::{AddAction, FileEntry}; + use crate::txlog::actions::{AddAction, FileEntry, ManifestInfo}; use std::collections::HashMap; fn make_entry(path: &str) -> FileEntry { @@ -314,100 +176,6 @@ mod tests { assert_eq!(result[0].added_at_version, 7); } - // ======================================================================== - // Selective compaction tests - // ======================================================================== - - fn make_manifest_info(path: &str, count: i64, bounds: Option) -> ManifestInfo { - ManifestInfo { - path: path.to_string(), - file_count: count, - partition_bounds: bounds, - ..Default::default() - } - } - - fn make_bounds(col: &str, min: &str, max: &str) -> PartitionBounds { - let mut min_values = HashMap::new(); - min_values.insert(col.to_string(), min.to_string()); - let mut max_values = HashMap::new(); - max_values.insert(col.to_string(), max.to_string()); - PartitionBounds { min_values, max_values } - } - - #[test] - fn test_distribute_tombstones_to_manifests_empty() { - let manifests = vec![make_manifest_info("m1.avro", 100, None)]; - let tombstones = HashSet::new(); - let result = distribute_tombstones_to_manifests(&manifests, &tombstones, &[]); - assert_eq!(result[0].tombstone_count, 0); - assert_eq!(result[0].live_entry_count, 100); - } - - #[test] - fn test_distribute_tombstones_with_partition_bounds() { - let manifests = vec![ - make_manifest_info("m-2023.avro", 100, - Some(make_bounds("year", "2023", "2023"))), - make_manifest_info("m-2024.avro", 100, - Some(make_bounds("year", "2024", "2024"))), - ]; - let mut tombstones = HashSet::new(); - tombstones.insert("year=2024/file1.split".to_string()); - tombstones.insert("year=2024/file2.split".to_string()); - tombstones.insert("year=2023/file3.split".to_string()); - - let result = distribute_tombstones_to_manifests(&manifests, &tombstones, &["year".to_string()]); - assert_eq!(result[0].tombstone_count, 1); // year=2023 - assert_eq!(result[1].tombstone_count, 2); // year=2024 - } - - #[test] - fn test_selective_partition() { - let manifests = vec![ - ManifestInfo { tombstone_count: 0, live_entry_count: 100, file_count: 100, ..Default::default() }, - ManifestInfo { tombstone_count: 50, live_entry_count: 50, file_count: 100, ..Default::default() }, - ManifestInfo { tombstone_count: 5, live_entry_count: 95, file_count: 100, ..Default::default() }, - ]; - - let (keep, rewrite) = selective_partition(&manifests, 0.10); - assert_eq!(keep.len(), 2); // 0% and 5% - assert_eq!(rewrite.len(), 1); // 50% - } - - #[test] - fn test_selective_compaction_beneficial() { - let keep = vec![ - ManifestInfo { file_count: 1000, ..Default::default() }, - ]; - let rewrite = vec![ - ManifestInfo { file_count: 100, ..Default::default() }, - ]; - // 1000 / 1100 = 90.9% savings — beneficial - assert!(is_selective_compaction_beneficial(&keep, &rewrite)); - } - - #[test] - fn test_selective_compaction_not_beneficial_all_dirty() { - let keep: Vec = vec![]; - let rewrite = vec![ManifestInfo { file_count: 100, ..Default::default() }]; - assert!(!is_selective_compaction_beneficial(&keep, &rewrite)); - } - - #[test] - fn test_extract_partition_values() { - let values = extract_partition_values("year=2024/month=01/file.split", &[]); - assert_eq!(values.get("year"), Some(&"2024".to_string())); - assert_eq!(values.get("month"), Some(&"01".to_string())); - assert_eq!(values.len(), 2); - } - - #[test] - fn test_extract_partition_values_no_partitions() { - let values = extract_partition_values("plain/file.split", &[]); - assert!(values.is_empty()); - } - #[test] fn test_needs_compaction_high_tombstones() { use crate::txlog::actions::StateManifest; diff --git a/src/test/java/io/indextables/jni/txlog/TransactionLogIntegrationTest.java b/src/test/java/io/indextables/jni/txlog/TransactionLogIntegrationTest.java index 612f3fd5..e8197858 100644 --- a/src/test/java/io/indextables/jni/txlog/TransactionLogIntegrationTest.java +++ b/src/test/java/io/indextables/jni/txlog/TransactionLogIntegrationTest.java @@ -1677,10 +1677,16 @@ void testPurgeWithAgingReproduction() throws Exception { ", manifestPaths=" + snap2.getManifestPaths().size() + ", postCheckpointPaths=" + snap2.getPostCheckpointPaths().size()); - // Step 5: Write (append) AFTER purge — this is what triggers the bug - // Use the exact config from the Scala test: checkpoint_interval=10, cache.ttl.ms=300000 + // Step 5: Write (append) AFTER purge — this is what triggers the bug. + // checkpoint_interval is honored as a real interval (checkpoint every N + // commits, version % N == 0), so we set it to 1 here to force an + // auto-checkpoint on THIS append — step 6b/6c below specifically verify that + // the auto-checkpoint carries forward metadata and file entries from the + // previous checkpoint. (The reproduced Scala test used interval=10, which + // under modulo semantics would not checkpoint at version 13 and so would + // not exercise the carry-forward path this test targets.) Map writeConfig = new HashMap<>(config); - writeConfig.put("checkpoint_interval", "10"); + writeConfig.put("checkpoint_interval", "1"); writeConfig.put("cache.ttl.ms", "300000"); String appendAdds = MAPPER.writeValueAsString(List.of( makeAddAction("split-13.split", 1000, 1)));