From c17588044f7b8cf0ea592daffa756ac32f800f90 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 7 Jun 2026 12:56:36 +0000 Subject: [PATCH] optimize: ZOrder idempotence guard to stop re-churning unchanged partitions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit delta-rs's ZOrder planner has no idempotence guard — no size check, no single-file-bin skip, no "already sorted" detection — so the full optimize job rewrote every file in today + yesterday + 2-days-ago into new (cold, multipart) object keys on every 30-min run, even for sealed days that hadn't changed. PR #39 added cache warming to soften the resulting cold-starts, but the cheaper fix is to not do the redundant rewrites in the first place. optimize_table now snapshots each window partition's live file set and skips any partition whose set is unchanged since the last successful optimize; `today` is always processed (the growing leading edge). Sealed days are z-ordered once, then skipped — so #39's warm/evict has far less cold churn to chase. The new file set is recorded in an in-memory per-table map (zorder_filesets); a restart re-z-orders each in-window partition once, harmlessly. Composes with #39: this narrows what gets rewritten; #39 still warms whatever actually was. New metrics optimize.partitions_rewritten / partitions_skipped expose the churn-avoided signal (after-the-fact monitoring, no telemetry pipeline needed). Unit tests cover the fileset grouping and change detection. https://claude.ai/code/session_01PLU62s7eRhSMMUXEHSWRCx --- src/database.rs | 157 +++++++++++++++++++++++++++++++++++++++++++----- src/metrics.rs | 11 ++++ 2 files changed, 152 insertions(+), 16 deletions(-) diff --git a/src/database.rs b/src/database.rs index 5f0e4181..c148193c 100644 --- a/src/database.rs +++ b/src/database.rs @@ -49,6 +49,10 @@ pub type UnifiedTables = Arc>>>>; // Key: (project_id, table_name) -> DeltaTable pub type CustomProjectTables = Arc>>>>; +// Per-table (keyed by storage URL), per-date set of live file URIs at the last +// successful z-order optimize. Backs the ZOrder idempotence guard. +type ZOrderFilesets = Arc>>>>; + /// Get a Delta table from custom project tables by project_id and table_name pub async fn get_custom_delta_table(custom_tables: &CustomProjectTables, project_id: &str, table_name: &str) -> Option>> { custom_tables.read().await.get(&(project_id.to_string(), table_name.to_string())).cloned() @@ -451,6 +455,15 @@ pub struct Database { buffered_layer: Option>, tantivy_search: Option>, tantivy_indexer: Option>, + /// Per-table, per-date set of live file URIs as of the last successful full + /// (z-order) optimize. delta-rs's ZOrder planner has no idempotence guard — + /// it rewrites every file in the window on every run, even sealed days that + /// didn't change, minting cold multipart objects that cold-start the + /// object-store cache (which PR #39 then has to re-warm). This lets + /// `optimize_table` skip a sealed partition whose file set is unchanged. + /// Keyed by table storage URL (unique per physical table). In-memory only: + /// a restart re-z-orders each partition once, which is harmless. + zorder_filesets: ZOrderFilesets, } impl Database { @@ -740,6 +753,7 @@ impl Database { buffered_layer: None, tantivy_search: None, tantivy_indexer: None, + zorder_filesets: Arc::new(RwLock::new(HashMap::new())), }; Ok(db) @@ -2154,34 +2168,73 @@ impl Database { pub async fn optimize_table(&self, table_ref: &Arc>, table_name: &str, _target_size: Option) -> Result<()> { let start_time = std::time::Instant::now(); let window_hours = self.config.maintenance.timefusion_optimize_window_hours.max(1); - info!("Starting Delta table optimization with Z-ordering (last {} hours)", window_hours); let table_clone = { let table = table_ref.read().await; table.clone() }; + // Candidate date partitions in the window (today .. today-num_days). + let now = Utc::now(); + let today = now.date_naive(); + let num_days = (window_hours / 24).max(1); + let window_dates: Vec = (0..=num_days).map(|days_ago| (now - chrono::Duration::days(days_ago as i64)).date_naive()).collect(); + + // Snapshot the current live file set once: drives both the ZOrder + // idempotence guard (below) and PR #39's warm/evict (`pre_uris`). + let all_uris: Vec = table_clone.get_file_uris().map(|it| it.collect()).unwrap_or_default(); + let table_url = table_clone.table_url().to_string(); + let current = Self::filesets_for_dates(&all_uris, &window_dates); + // Pre-state file set, used to derive the files this optimize *adds* // (to warm) and *removes* (to evict) — see warm/evict_cache_for_uris. let track_files = self.config.maintenance.timefusion_warm_after_compaction || self.config.maintenance.timefusion_evict_after_compaction; - let pre_uris: std::collections::HashSet = if track_files { - table_clone.get_file_uris().map(|it| it.collect()).unwrap_or_default() - } else { - Default::default() - }; + let pre_uris: std::collections::HashSet = if track_files { all_uris.iter().cloned().collect() } else { Default::default() }; let target_size = self.config.parquet.timefusion_optimize_target_size; - // Generate partition filters for each date in the configurable window - let now = Utc::now(); - let num_days = (window_hours / 24).max(1); - let partition_filters: Vec = (0..=num_days) - .filter_map(|days_ago| { - let date = (now - chrono::Duration::days(days_ago as i64)).date_naive(); - PartitionFilter::try_from(("date", "=", date.to_string().as_str())).ok() - }) - .collect(); - info!("Optimizing files from {} date partitions", partition_filters.len()); + // delta-rs ZOrder has NO idempotence guard (unlike Compact it does no + // size / single-file / already-sorted check): it rewrites every file in + // the selected partitions on every run, even sealed days that didn't + // change — and PR #39 then has to re-warm all those cold rewrites. Skip + // any partition whose live file set is identical to the last successful + // optimize. `today` is always processed (growing leading edge). + let kept_dates: Vec = { + let guard = self.zorder_filesets.read().await; + let prev = guard.get(&table_url); + window_dates + .iter() + .filter(|d| match current.get(*d) { + None => false, + Some(cur) if cur.is_empty() => false, + Some(cur) => **d == today || prev.and_then(|m| m.get(*d)).map(|p| p != cur).unwrap_or(true), + }) + .copied() + .collect() + }; + let skipped = window_dates.len().saturating_sub(kept_dates.len()); + + if kept_dates.is_empty() { + info!( + "optimize: table={} all {} window partitions unchanged since last run — skipping (cache churn avoided)", + table_name, + window_dates.len() + ); + crate::metrics::record_optimize_partitions(0, skipped as u64); + return Ok(()); + } + + info!( + "Starting optimize (z-order): table={} rewriting {} of {} window partitions, skipping {} unchanged (last {}h)", + table_name, + kept_dates.len(), + window_dates.len(), + skipped, + window_hours + ); + + let partition_filters: Vec = + kept_dates.iter().filter_map(|d| PartitionFilter::try_from(("date", "=", d.to_string().as_str())).ok()).collect(); let schema = get_schema(table_name).unwrap_or_else(get_default_schema); // Full Z-order optimize runs every 30 min over a 48h window — promote @@ -2211,6 +2264,22 @@ impl Database { match optimize_result { Ok((new_table, metrics)) => { + // Record the post-commit file set for the partitions we + // rewrote so the next run skips them if nothing changes. Done + // before the min_files early-return so state stays consistent + // even when we don't adopt the new handle (delta-rs has already + // committed the rewrite by this point regardless). + { + let new_uris: Vec = new_table.get_file_uris().map(|it| it.collect()).unwrap_or_default(); + let new_sets = Self::filesets_for_dates(&new_uris, &kept_dates); + let mut guard = self.zorder_filesets.write().await; + let entry = guard.entry(table_url.clone()).or_default(); + for d in &kept_dates { + entry.insert(*d, new_sets.get(d).cloned().unwrap_or_default()); + } + } + crate::metrics::record_optimize_partitions(kept_dates.len() as u64, skipped as u64); + let min_files = self.config.maintenance.timefusion_compact_min_files; if metrics.total_considered_files < min_files { debug!( @@ -2270,6 +2339,24 @@ impl Database { } } + /// Group live file URIs by their `date=YYYY-MM-DD` Hive partition, for the + /// given dates only. URIs not matching any of `dates` are ignored. Every + /// requested date gets an entry (possibly empty) so the idempotence guard + /// can tell "no files" from "not looked at". + fn filesets_for_dates(uris: &[String], dates: &[chrono::NaiveDate]) -> HashMap> { + let markers: Vec<(chrono::NaiveDate, String)> = dates.iter().map(|d| (*d, format!("date={d}"))).collect(); + let mut out: HashMap> = dates.iter().map(|d| (*d, std::collections::HashSet::new())).collect(); + for uri in uris { + for (d, marker) in &markers { + if uri.contains(marker) { + out.get_mut(d).expect("date pre-seeded").insert(uri.clone()); + break; + } + } + } + out + } + /// Rewrites a date partition at a higher ZSTD level using Z-order (or /// Compact if no z_order_columns). Skips partitions whose probe file /// already advertises a tier `>= target_level` via Parquet footer KV @@ -3629,6 +3716,44 @@ mod tests { assert!(parsed[1..].iter().all(|p| p.is_none())); } + /// `filesets_for_dates` buckets URIs by their `date=` partition and + /// pre-seeds every requested date (so the guard can tell "empty" from + /// "absent"). URIs outside the requested dates are dropped. + #[test] + fn filesets_for_dates_groups_by_partition() { + use std::collections::HashSet; + let d0 = chrono::NaiveDate::from_ymd_opt(2026, 6, 6).unwrap(); + let d1 = chrono::NaiveDate::from_ymd_opt(2026, 6, 5).unwrap(); + let uris = vec![ + "s3://b/t/date=2026-06-06/part-a.parquet".to_string(), + "s3://b/t/date=2026-06-06/part-b.parquet".to_string(), + "s3://b/t/date=2026-06-05/part-c.parquet".to_string(), + "s3://b/t/date=2026-06-01/part-x.parquet".to_string(), // outside window + ]; + let sets = Database::filesets_for_dates(&uris, &[d0, d1]); + assert_eq!(sets[&d0].len(), 2); + assert_eq!(sets[&d1], HashSet::from(["s3://b/t/date=2026-06-05/part-c.parquet".to_string()])); + // A date with no files is still present (empty), not missing. + let d2 = chrono::NaiveDate::from_ymd_opt(2026, 6, 4).unwrap(); + let sets = Database::filesets_for_dates(&uris, &[d2]); + assert!(sets[&d2].is_empty()); + } + + /// Two identical file sets compare equal (→ partition skipped); adding a + /// file makes them differ (→ partition re-optimized). This is the core of + /// the ZOrder idempotence guard. + #[test] + fn filesets_equal_only_when_unchanged() { + let d = chrono::NaiveDate::from_ymd_opt(2026, 6, 6).unwrap(); + let base = vec!["s3://b/t/date=2026-06-06/a.parquet".to_string()]; + let plus = vec!["s3://b/t/date=2026-06-06/a.parquet".to_string(), "s3://b/t/date=2026-06-06/b.parquet".to_string()]; + let a = Database::filesets_for_dates(&base, &[d]); + let b = Database::filesets_for_dates(&base, &[d]); + let c = Database::filesets_for_dates(&plus, &[d]); + assert_eq!(a[&d], b[&d]); + assert_ne!(a[&d], c[&d]); + } + /// Helper function to extract string value from array column, handling different string array types fn get_str(array: &dyn Array, idx: usize) -> String { use datafusion::arrow::array::{LargeStringArray, StringArray, StringViewArray}; diff --git a/src/metrics.rs b/src/metrics.rs index 1cf2086c..aa60bcfd 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -71,6 +71,8 @@ counter_registry! { tantivy_prefilter_errors => "timefusion.tantivy.prefilter_errors": "Tantivy lookups that errored (S3 down, parse failure, etc.)", tantivy_build_failures => "timefusion.tantivy.build_failures": "Post-flush tantivy index builds that errored — accumulating drift means queries silently fall back to UDF scan", dedup_dropped_rows => "timefusion.flush.dedup_dropped_rows": "Rows collapsed by per-table dedup_keys (last-write-wins) before Delta commit", + optimize_partitions_rewritten => "timefusion.optimize.partitions_rewritten": "Date partitions rewritten by full (z-order) optimize", + optimize_partitions_skipped => "timefusion.optimize.partitions_skipped": "Date partitions skipped by full optimize because their file set was unchanged since the last run (cache churn avoided)", } pub fn registry() -> Option<&'static MetricsRegistry> { @@ -275,3 +277,12 @@ pub fn record_dedup_dropped(rows: u64) { m.dedup_dropped_rows.add(rows, &[]); } } + +/// Record one full-optimize run's idempotence split: how many window partitions +/// were rewritten vs skipped as unchanged (the cache-churn-avoided signal). +pub fn record_optimize_partitions(rewritten: u64, skipped: u64) { + if let Some(m) = METRICS.get() { + m.optimize_partitions_rewritten.add(rewritten, &[]); + m.optimize_partitions_skipped.add(skipped, &[]); + } +}