1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
use std::collections::HashSet;
use futures::{StreamExt, stream};
use object_store::ObjectStore;
use object_store::path::Path as ObjectStorePath;
use sea_query::{Query, UnionType};
use super::DryRun;
use crate::spec::*;
use crate::{Ducklake, DucklakeResult, io};
/* ----------------------------------------- PUBLIC API ---------------------------------------- */
impl Ducklake {
/// Delete files in the data directory that are not referenced by any snapshot.
///
/// Orphaned files are files that physically exist below the catalog's data path but are not
/// referenced by any data file, delete file, or file scheduled for deletion. Such files can
/// occur, for example, when a writer crashes after writing a file but before committing the
/// corresponding catalog entry.
///
/// The `dry_run` flag allows understanding which files would be deleted upon execution without
/// actually deleting them.
///
/// Returns the fully-qualified paths of the files that were deleted (or that would be deleted
/// when `dry_run` is [`DryRun::Yes`]).
pub async fn delete_orphaned_files(&self, dry_run: DryRun) -> DucklakeResult<Vec<String>> {
let interval = self.conn.metadata().delete_older_than();
let timestamp = chrono::Utc::now() - interval.months - interval.delta;
self.delete_orphaned_files_filtered(Some(timestamp), dry_run)
.await
}
/// Delete orphaned files last modified before a specific timestamp.
///
/// The functionality matches [`Ducklake::delete_orphaned_files`] for a predefined timestamp.
pub async fn delete_orphaned_files_older_than(
&self,
timestamp: chrono::DateTime<chrono::Utc>,
dry_run: DryRun,
) -> DucklakeResult<Vec<String>> {
self.delete_orphaned_files_filtered(Some(timestamp), dry_run)
.await
}
/// Delete all orphaned files regardless of their last modification time.
///
/// The functionality matches [`Ducklake::delete_orphaned_files`] but ignores the
/// `delete_older_than` configuration.
pub async fn delete_all_orphaned_files(&self, dry_run: DryRun) -> DucklakeResult<Vec<String>> {
self.delete_orphaned_files_filtered(None, dry_run).await
}
async fn delete_orphaned_files_filtered(
&self,
max_age: Option<chrono::DateTime<chrono::Utc>>,
dry_run: DryRun,
) -> DucklakeResult<Vec<String>> {
let data_path = self.conn.metadata().data_path();
// 1) Collect all object store locations referenced by the catalog.
let referenced = self.collect_referenced_locations(&data_path).await?;
// 2) List all files below the data path and find the orphaned ones.
let resolved_data_path = data_path.resolve()?;
let store = resolved_data_path.object_store(Some(self.conn.storage_options().to_vec()));
let prefix = resolved_data_path.path();
let mut orphans = Vec::new();
let mut listing = store.list(Some(&prefix));
while let Some(meta) = listing.next().await {
let meta = meta?;
if referenced.contains(&meta.location) {
continue;
}
if let Some(max_age) = max_age
&& meta.last_modified >= max_age
{
continue;
}
orphans.push(meta.location);
}
// 3) Delete the orphaned files unless this is a dry run. We batch the deletes via
// `delete_stream` so the object store can use batch APIs where available, and tolerate
// files that are already gone to keep the operation idempotent.
if matches!(dry_run, DryRun::No) {
let locations = orphans.clone();
let mut deletion =
store.delete_stream(stream::iter(locations.into_iter().map(Ok)).boxed());
while let Some(result) = deletion.next().await {
match result {
Ok(_) | Err(object_store::Error::NotFound { .. }) => {}
Err(e) => return Err(e.into()),
}
}
}
Ok(orphans
.iter()
.map(|location| resolved_data_path.display_location(location))
.collect())
}
/// Collect the object store locations of all files referenced by the catalog, i.e. all data
/// files, delete files, and files already scheduled for deletion.
async fn collect_referenced_locations(
&self,
data_path: &io::DucklakePath,
) -> DucklakeResult<HashSet<ObjectStorePath>> {
let pool = self.conn.pool();
let snapshot = self.conn.latest_snapshot(false).await?;
let catalog = snapshot.catalog().await?;
// Fetch the file paths from all relevant catalog tables.
let data_file_query = Query::select()
.columns([
ducklake_data_file::Column::TableId,
ducklake_data_file::Column::Path,
ducklake_data_file::Column::PathIsRelative,
])
.from(ducklake_data_file::Table)
.union(
UnionType::All,
Query::select()
.columns([
ducklake_delete_file::Column::TableId,
ducklake_delete_file::Column::Path,
ducklake_delete_file::Column::PathIsRelative,
])
.from(ducklake_delete_file::Table)
.take(),
)
.take();
let table_files: Vec<(i64, String, bool)> = pool.fetch_all(&data_file_query).await?;
let scheduled_query = Query::select()
.columns([
ducklake_files_scheduled_for_deletion::Column::Path,
ducklake_files_scheduled_for_deletion::Column::PathIsRelative,
])
.from(ducklake_files_scheduled_for_deletion::Table)
.take();
let scheduled_files: Vec<(String, bool)> = pool.fetch_all(&scheduled_query).await?;
// Resolve all referenced file paths to object store locations.
let mut result = HashSet::new();
for (table_id, path, path_is_relative) in table_files {
let file_path = io::DucklakePath::new(&path, path_is_relative);
let path = catalog
.table(table_id)
.unwrap()
.data_path(data_path)
.join(&file_path);
result.insert(path.resolve()?.path());
}
// Files scheduled for deletion store paths relative to the catalog's data path.
for (path, path_is_relative) in scheduled_files {
let file_path = io::DucklakePath::new(&path, path_is_relative);
let path = data_path.join(&file_path);
result.insert(path.resolve()?.path());
}
Ok(result)
}
}