micromegas_analytics/lakehouse/
temp.rs

1use anyhow::{Context, Result};
2use chrono::{DateTime, Utc};
3use micromegas_ingestion::data_lake_connection::DataLakeConnection;
4use micromegas_tracing::prelude::*;
5use sqlx::Row;
6use std::sync::Arc;
7
8#[span_fn]
9async fn delete_expired_temporary_files_batch(
10    lake: &DataLakeConnection,
11    now: DateTime<Utc>,
12) -> Result<bool> {
13    let batch_size: i32 = 1000;
14    let mut tr = lake.db_pool.begin().await?;
15    let rows = instrument_named!(
16        sqlx::query(
17            "DELETE FROM temporary_files
18         WHERE file_path IN (
19             SELECT file_path FROM temporary_files
20             WHERE expiration < $1
21             LIMIT $2
22         )
23         RETURNING file_path;",
24        )
25        .bind(now)
26        .bind(batch_size)
27        .fetch_all(&mut *tr),
28        "sql_delete_expired_temporary_files_batch"
29    )
30    .await
31    .with_context(|| "deleting expired temporary files batch")?;
32
33    if rows.is_empty() {
34        return Ok(false);
35    }
36
37    let to_delete: Vec<String> = rows
38        .iter()
39        .map(|r| r.try_get("file_path"))
40        .collect::<Result<_, _>>()?;
41
42    for file_path in &to_delete {
43        debug!("deleting expired temporary file {file_path}");
44    }
45
46    lake.blob_storage.delete_batch(&to_delete).await?;
47    tr.commit().await?;
48    info!("deleted {} expired temporary files", to_delete.len());
49    Ok(true)
50}
51
52#[span_fn]
53pub async fn delete_expired_temporary_files(lake: Arc<DataLakeConnection>) -> Result<()> {
54    let now = Utc::now();
55    while delete_expired_temporary_files_batch(&lake, now).await? {}
56    Ok(())
57}
micromegas_analytics/lakehouse/temp.rs

micromegas_analytics/lakehouse/
temp.rs
