Skip to content

Commit f6f9c50

Browse files
parmesantparmesant
committed
x86 build + other fixes
1 parent b185b38 commit f6f9c50

6 files changed

Lines changed: 201 additions & 108 deletions

File tree

‎src/cli.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -341,14 +341,16 @@ pub struct Options {
341341
#[arg(
342342
long = "hot-tier-latest-minutes",
343343
env = "P_HOT_TIER_LATEST_MINUTES",
344+
value_parser = clap::value_parser!(u64).range(1..),
344345
default_value = "10",
345346
help = "Files whose timestamp is within the last N minutes are 'latest'; rest are 'historic'."
346347
)]
347-
pub hot_tier_latest_minutes: i64,
348+
pub hot_tier_latest_minutes: u64,
348349

349350
#[arg(
350351
long = "hot-tier-historic-sync-minutes",
351352
env = "P_HOT_TIER_HISTORIC_SYNC_MINUTES",
353+
value_parser = clap::value_parser!(u32).range(1..),
352354
default_value = "5",
353355
help = "Interval (minutes) at which the historic hot-tier sync runs."
354356
)]

‎src/hottier.rs‎

Lines changed: 101 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -221,31 +221,41 @@ impl HotTierManager {
221221
stream: &str,
222222
tenant_id: &Option<String>,
223223
) -> Result<(), HotTierError> {
224-
let mut entries = fs::read_dir(date_dir).await?;
225-
while let Some(entry) = entries.next_entry().await? {
226-
let p = entry.path();
227-
let Some(name_os) = p.file_name() else {
228-
continue;
229-
};
230-
let name = name_os.to_string_lossy();
231-
if name.ends_with(".partial") {
232-
let _ = fs::remove_file(&p).await;
233-
*partials_removed += 1;
234-
info!(
235-
stream = %stream,
236-
tenant = ?tenant_id,
237-
path = %p.display(),
238-
"reconcile: deleted partial orphan"
239-
);
240-
continue;
241-
}
242-
if name.ends_with(".manifest.json") {
243-
continue;
244-
}
245-
if !p.is_file() {
246-
continue;
224+
let mut stack: Vec<PathBuf> = vec![date_dir.clone()];
225+
while let Some(dir) = stack.pop() {
226+
let mut entries = fs::read_dir(&dir).await?;
227+
while let Some(entry) = entries.next_entry().await? {
228+
let p = entry.path();
229+
let ft = entry.file_type().await?;
230+
if ft.is_dir() {
231+
stack.push(p);
232+
continue;
233+
}
234+
let Some(name_os) = p.file_name() else {
235+
continue;
236+
};
237+
let name = name_os.to_string_lossy();
238+
if name.ends_with(".partial") {
239+
let _ = fs::remove_file(&p).await;
240+
*partials_removed += 1;
241+
info!(
242+
stream = %stream,
243+
tenant = ?tenant_id,
244+
path = %p.display(),
245+
"reconcile: deleted partial orphan"
246+
);
247+
continue;
248+
}
249+
if name.ends_with(".manifest.json") {
250+
continue;
251+
}
252+
if !ft.is_file() {
253+
continue;
254+
}
255+
if let Ok(rel) = p.strip_prefix(date_dir) {
256+
on_disk.insert(rel.to_string_lossy().into_owned());
257+
}
247258
}
248-
on_disk.insert(name.into_owned());
249259
}
250260
Ok(())
251261
}
@@ -275,8 +285,8 @@ impl HotTierManager {
275285
Err(_) => false,
276286
};
277287
if ok {
278-
if let Some(name) = local.file_name().and_then(|s| s.to_str()) {
279-
keep_names.insert(name.to_owned());
288+
if let Ok(rel) = local.strip_prefix(date_dir) {
289+
keep_names.insert(rel.to_string_lossy().into_owned());
280290
}
281291
*total_used += f.file_size;
282292
kept.push(f);
@@ -493,7 +503,7 @@ impl HotTierManager {
493503
'a: 'static,
494504
{
495505
let latest_min = PARSEABLE.options.hot_tier_latest_minutes;
496-
let historic_min = PARSEABLE.options.hot_tier_historic_sync_minutes.max(1);
506+
let historic_min = PARSEABLE.options.hot_tier_historic_sync_minutes;
497507
info!(
498508
latest_minutes = latest_min,
499509
historic_sync_minutes = historic_min,
@@ -516,6 +526,21 @@ impl HotTierManager {
516526
for stream in PARSEABLE.streams.list(&tenant_id) {
517527
if this.check_stream_hot_tier_exists(&stream, &tenant_id) {
518528
this.spawn_stream_tasks(stream, tenant_id.clone()).await;
529+
} else {
530+
// check for potential orphan directory on disk
531+
let path = if let Some(tenant_id) = tenant_id.as_ref() {
532+
self.hot_tier_path.join(tenant_id).join(stream)
533+
} else {
534+
self.hot_tier_path.join(stream)
535+
};
536+
if path.exists() {
537+
// delete this entire folder as stream meta says no hottier for stream
538+
if let Err(e) = fs::remove_dir_all(&path).await {
539+
tracing::error!(
540+
"Unable to remove orphaned hottier dir- `{path:?}` with error- {e}"
541+
);
542+
};
543+
}
519544
}
520545
}
521546
}
@@ -538,9 +563,8 @@ impl HotTierManager {
538563
}
539564

540565
let latest_interval = Duration::from_secs(60);
541-
let historic_interval = Duration::from_secs(
542-
PARSEABLE.options.hot_tier_historic_sync_minutes.max(1) as u64 * 60,
543-
);
566+
let historic_interval =
567+
Duration::from_secs(PARSEABLE.options.hot_tier_historic_sync_minutes as u64 * 60);
544568

545569
info!(stream = %stream, tenant = ?tenant_id, "spawning per-stream hot tier tasks");
546570

@@ -658,8 +682,9 @@ impl HotTierManager {
658682
return Ok(());
659683
}
660684

661-
let latest_minutes = PARSEABLE.options.hot_tier_latest_minutes.max(0);
662-
let cutoff = chrono::Utc::now().naive_utc() - chrono::Duration::minutes(latest_minutes);
685+
let latest_minutes = PARSEABLE.options.hot_tier_latest_minutes;
686+
let cutoff = chrono::Utc::now().naive_utc()
687+
- chrono::Duration::minutes(latest_minutes.try_into().unwrap());
663688

664689
// Build flat work list: (date, file, local_path) ordered latest-date-first,
665690
// and within a date by descending file_path (matches old `pop()` order).
@@ -717,10 +742,7 @@ impl HotTierManager {
717742
);
718743

719744
let state = self.get_or_load_state(stream, tenant_id).await?;
720-
let concurrency = PARSEABLE
721-
.options
722-
.hot_tier_files_per_stream_concurrency
723-
.max(4);
745+
let concurrency = PARSEABLE.options.hot_tier_files_per_stream_concurrency;
724746

725747
// Reservation failure (out of disk + nothing to evict) is sticky:
726748
// once one file can't be placed, no subsequent file will fit either.
@@ -1150,9 +1172,23 @@ impl HotTierManager {
11501172
let mut delete_successful = false;
11511173
let mut freed_total: u64 = 0;
11521174
let dates = self.fetch_hot_tier_dates(stream, tenant_id).await?;
1175+
if dates.is_empty() {
1176+
info!(
1177+
stream = %stream,
1178+
tenant = ?tenant_id,
1179+
"eviction: no date dirs found, nothing to evict"
1180+
);
1181+
}
11531182
'loop_dates: for date in dates {
11541183
let path = self.get_stream_path_for_date(stream, &date, tenant_id);
11551184
if !path.exists() {
1185+
info!(
1186+
stream = %stream,
1187+
tenant = ?tenant_id,
1188+
date = %date,
1189+
path = %path.display(),
1190+
"eviction: date path missing, skipping"
1191+
);
11561192
continue;
11571193
}
11581194

@@ -1164,11 +1200,27 @@ impl HotTierManager {
11641200
.to_string_lossy()
11651201
.ends_with(".manifest.json")
11661202
});
1203+
if manifest_files.is_empty() {
1204+
info!(
1205+
stream = %stream,
1206+
tenant = ?tenant_id,
1207+
date = %date,
1208+
path = %path.display(),
1209+
"eviction: no .manifest.json files in date dir"
1210+
);
1211+
continue;
1212+
}
11671213
for manifest_file in manifest_files {
11681214
let file = fs::read(manifest_file.path()).await?;
11691215
let mut manifest: Manifest = serde_json::from_slice(&file)?;
11701216

11711217
if manifest.files.is_empty() {
1218+
info!(
1219+
stream = %stream,
1220+
tenant = ?tenant_id,
1221+
manifest = %manifest_file.path().display(),
1222+
"eviction: manifest has zero file entries"
1223+
);
11721224
continue;
11731225
}
11741226

@@ -1182,7 +1234,18 @@ impl HotTierManager {
11821234
let first_file_path = self.hot_tier_path.join(&first_file.file_path);
11831235
let minute_to_delete = first_file_path.parent().unwrap();
11841236

1185-
if minute_to_delete.exists() {
1237+
if !minute_to_delete.exists() {
1238+
info!(
1239+
stream = %stream,
1240+
tenant = ?tenant_id,
1241+
manifest = %manifest_file.path().display(),
1242+
first_file = %first_file.file_path,
1243+
minute = %minute_to_delete.display(),
1244+
"eviction: minute dir referenced by manifest does not exist on disk"
1245+
);
1246+
continue;
1247+
}
1248+
{
11861249
if let (Some(download_date_time), Some(delete_date_time)) = (
11871250
extract_datetime(download_file_path.to_str().unwrap()),
11881251
extract_datetime(first_file_path.to_str().unwrap()),
@@ -1244,7 +1307,7 @@ impl HotTierManager {
12441307
new_available = stream_hot_tier.available_size,
12451308
"evicted"
12461309
);
1247-
if stream_hot_tier.available_size <= parquet_file_size {
1310+
if stream_hot_tier.available_size < parquet_file_size {
12481311
continue;
12491312
} else {
12501313
break 'loop_dates;

‎src/storage/azure_blob.rs‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -224,7 +224,10 @@ impl BlobStore {
224224
.await
225225
{
226226
Ok(()) => {
227-
tokio::fs::rename(&partial, &write_path).await?;
227+
if let Err(e) = tokio::fs::rename(&partial, &write_path).await {
228+
let _ = tokio::fs::remove_file(&partial).await;
229+
return Err(e.into());
230+
}
228231
Ok(())
229232
}
230233
Err(e) => {
@@ -255,11 +258,8 @@ impl BlobStore {
255258
file.set_len(total).await?;
256259
let std_file = Arc::new(file.into_std().await);
257260

258-
let chunk = PARSEABLE
259-
.options
260-
.hot_tier_download_chunk_size
261-
.max(8 * 1024 * 1024);
262-
let concurrency = PARSEABLE.options.hot_tier_download_concurrency.max(6);
261+
let chunk = PARSEABLE.options.hot_tier_download_chunk_size;
262+
let concurrency = PARSEABLE.options.hot_tier_download_concurrency;
263263
let ranges: Vec<Range<u64>> = (0..total)
264264
.step_by(chunk as usize)
265265
.map(|s| s..(s + chunk).min(total))
@@ -281,8 +281,7 @@ impl BlobStore {
281281
let bytes = client.get_range(&src, r.clone()).await?;
282282
let offset = r.start;
283283
tokio::task::spawn_blocking(move || -> std::io::Result<()> {
284-
use std::os::unix::fs::FileExt;
285-
std_file.write_all_at(&bytes, offset)
284+
crate::storage::write_all_at(&std_file, &bytes, offset)
286285
})
287286
.await
288287
.map_err(|e| ObjectStorageError::Custom(format!("join: {e}")))??;

‎src/storage/gcs.rs‎

Lines changed: 28 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -190,7 +190,10 @@ impl Gcs {
190190
.await
191191
{
192192
Ok(()) => {
193-
tokio::fs::rename(&partial, &write_path).await?;
193+
if let Err(e) = tokio::fs::rename(&partial, &write_path).await {
194+
let _ = tokio::fs::remove_file(&partial).await;
195+
return Err(e.into());
196+
}
194197
Ok(())
195198
}
196199
Err(e) => {
@@ -221,11 +224,8 @@ impl Gcs {
221224
file.set_len(total).await?;
222225
let std_file = Arc::new(file.into_std().await);
223226

224-
let chunk = PARSEABLE
225-
.options
226-
.hot_tier_download_chunk_size
227-
.max(8 * 1024 * 1024);
228-
let concurrency = PARSEABLE.options.hot_tier_download_concurrency.max(16);
227+
let chunk = PARSEABLE.options.hot_tier_download_chunk_size;
228+
let concurrency = PARSEABLE.options.hot_tier_download_concurrency;
229229
let ranges: Vec<Range<u64>> = (0..total)
230230
.step_by(chunk as usize)
231231
.map(|s| s..(s + chunk).min(total))
@@ -234,32 +234,29 @@ impl Gcs {
234234
let client = self.client.clone();
235235
let semaphore = Arc::new(tokio::sync::Semaphore::new(concurrency));
236236

237-
let mut handles = Vec::with_capacity(ranges.len());
238-
for r in ranges {
239-
let client = client.clone();
240-
let src = src.clone();
241-
let std_file = std_file.clone();
242-
let semaphore = semaphore.clone();
243-
handles.push(tokio::spawn(async move {
244-
let _permit = semaphore
245-
.acquire_owned()
237+
futures::stream::iter(ranges)
238+
.map(|r| {
239+
let client = client.clone();
240+
let src = src.clone();
241+
let std_file = std_file.clone();
242+
let semaphore = semaphore.clone();
243+
async move {
244+
let _permit = semaphore.acquire_owned().await.map_err(|e| {
245+
ObjectStorageError::Custom(format!("semaphore closed: {e}"))
246+
})?;
247+
let bytes = client.get_range(&src, r.clone()).await?;
248+
let offset = r.start;
249+
tokio::task::spawn_blocking(move || -> std::io::Result<()> {
250+
crate::storage::write_all_at(&std_file, &bytes, offset)
251+
})
246252
.await
247-
.map_err(|e| ObjectStorageError::Custom(format!("semaphore closed: {e}")))?;
248-
let bytes = client.get_range(&src, r.clone()).await?;
249-
let offset = r.start;
250-
tokio::task::spawn_blocking(move || -> std::io::Result<()> {
251-
use std::os::unix::fs::FileExt;
252-
std_file.write_all_at(&bytes, offset)
253-
})
254-
.await
255-
.map_err(|e| ObjectStorageError::Custom(format!("join: {e}")))??;
256-
Ok::<_, ObjectStorageError>(())
257-
}));
258-
}
259-
for h in handles {
260-
h.await
261-
.map_err(|e| ObjectStorageError::Custom(format!("join error: {e}")))??;
262-
}
253+
.map_err(|e| ObjectStorageError::Custom(format!("join: {e}")))??;
254+
Ok::<_, ObjectStorageError>(())
255+
}
256+
})
257+
.buffer_unordered(concurrency)
258+
.try_collect::<Vec<_>>()
259+
.await?;
263260

264261
let std_file_sync = std_file.clone();
265262
tokio::task::spawn_blocking(move || std_file_sync.sync_all())

0 commit comments

Comments
 (0)