feat: Hard delete database and table (#26553)

* feat: Hard delete database and table

* feat: Enable hard-deletion in OSS

Author: stuartcarnie

influxdb3/src/commands/delete.rs

@@ -1,5 +1,6 @@
 use super::common::InfluxDb3Config;
 use influxdb3_client::Client;
+use influxdb3_types::http::HardDeletionTime;
 use secrecy::ExposeSecret;
 use secrecy::Secret;
 use std::error::Error;
@@ -115,6 +116,11 @@ pub struct DatabaseConfig {
     #[clap(env = "INFLUXDB3_DATABASE_NAME", required = true)]
     pub database_name: String,
 
+    /// When to perform hard deletion (never/now/default/timestamp).
+    /// Examples: 'never', 'now', 'default', '2024-06-18T10:30:00Z'
+    #[clap(long = "hard-delete", value_name = "WHEN")]
+    pub hard_delete: Option<String>,
+
     /// An optional arg to use a custom ca for useful for testing with self signed certs
     #[clap(long = "tls-ca", env = "INFLUXDB3_TLS_CA")]
     ca_cert: Option<PathBuf>,
@@ -164,6 +170,11 @@ pub struct TableConfig {
     /// The name of the table to be deleted
     table_name: String,
 
+    /// When to perform hard deletion (never/now/default/timestamp).
+    /// Examples: 'never', 'now', 'default', '2024-06-18T10:30:00Z'
+    #[clap(long = "hard-delete", value_name = "WHEN")]
+    hard_delete: Option<String>,
+
     /// An optional arg to use a custom ca for useful for testing with self signed certs
     #[clap(long = "tls-ca", env = "INFLUXDB3_TLS_CA")]
     ca_cert: Option<PathBuf>,
@@ -211,10 +222,26 @@ pub struct TokenConfig {
     ca_cert: Option<PathBuf>,
 }
 
+fn parse_hard_delete_time(value: Option<String>) -> Option<HardDeletionTime> {
+    match value {
+        None => None,
+        Some(s) => match s.to_lowercase().as_str() {
+            "never" => Some(HardDeletionTime::Never),
+            "now" => Some(HardDeletionTime::Now),
+            "default" => Some(HardDeletionTime::Default),
+            _ => Some(HardDeletionTime::Timestamp(s)),
+        },
+    }
+}
+
 pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
     let client = config.get_client()?;
     match config.cmd {
-        SubCommand::Database(DatabaseConfig { database_name, .. }) => {
+        SubCommand::Database(DatabaseConfig {
+            database_name,
+            hard_delete,
+            ..
+        }) => {
             println!(
                 "Are you sure you want to delete {:?}? Enter 'yes' to confirm",
                 database_name
@@ -224,7 +251,18 @@ pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
             if confirmation.trim() != "yes" {
                 println!("Cannot delete database without confirmation");
             } else {
-                client.api_v3_configure_db_delete(&database_name).await?;
+                let hard_delete_time = parse_hard_delete_time(hard_delete);
+
+                if hard_delete_time.is_some() {
+                    client
+                        .api_v3_configure_db_delete_with_hard_delete(
+                            &database_name,
+                            hard_delete_time,
+                        )
+                        .await?;
+                } else {
+                    client.api_v3_configure_db_delete(&database_name).await?;
+                }
 
                 println!("Database {:?} deleted successfully", &database_name);
             }
@@ -256,6 +294,7 @@ pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
         SubCommand::Table(TableConfig {
             influxdb3_config: InfluxDb3Config { database_name, .. },
             table_name,
+            hard_delete,
             ..
         }) => {
             println!(
@@ -267,9 +306,21 @@ pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
             if confirmation.trim() != "yes" {
                 println!("Cannot delete table without confirmation");
             } else {
-                client
-                    .api_v3_configure_table_delete(&database_name, &table_name)
-                    .await?;
+                let hard_delete_time = parse_hard_delete_time(hard_delete);
+
+                if hard_delete_time.is_some() {
+                    client
+                        .api_v3_configure_table_delete_with_hard_delete(
+                            &database_name,
+                            &table_name,
+                            hard_delete_time,
+                        )
+                        .await?;
+                } else {
+                    client
+                        .api_v3_configure_table_delete(&database_name, &table_name)
+                        .await?;
+                }
 
                 println!(
                     "Table {:?}.{:?} deleted successfully",

influxdb3/src/commands/serve.rs

@@ -41,15 +41,15 @@ use influxdb3_telemetry::{
 };
 use influxdb3_wal::{Gen1Duration, WalConfig};
 use influxdb3_write::{
-    WriteBuffer,
+    WriteBuffer, deleter,
     persister::Persister,
     write_buffer::{
         WriteBufferImpl, WriteBufferImplArgs, check_mem_and_force_snapshot_loop,
         persisted_files::PersistedFiles,
     },
 };
 use iox_query::exec::{DedicatedExecutor, Executor, ExecutorConfig, PerQueryMemoryPoolConfig};
-use iox_time::SystemProvider;
+use iox_time::{SystemProvider, TimeProvider};
 use metric::U64Gauge;
 use object_store::ObjectStore;
 use object_store_metrics::ObjectStoreMetrics;
@@ -455,6 +455,16 @@ pub struct Config {
         default_value_t = Catalog::DEFAULT_HARD_DELETE_DURATION.into(),
     )]
     pub hard_delete_default_duration: humantime::Duration,
+
+    /// Grace period for hard deleted databases and tables before they are removed permanently from
+    /// the catalog.
+    #[clap(
+        long = "delete-grace-period",
+        env = "INFLUXDB3_DELETE_GRACE_PERIOD",
+        default_value = "24h",
+        action
+    )]
+    pub delete_grace_period: humantime::Duration,
 }
 
 /// The minimum version of TLS to use for InfluxDB
@@ -506,6 +516,9 @@ impl FromStr for MemorySizeMb {
         let num_bytes = if s.contains("%") {
             let mem_size = MemorySize::from_str(s)?;
             mem_size.bytes()
+        } else if let Some(suffix) = s.strip_suffix('b') {
+            usize::from_str(suffix)
+                .map_err(|_| "failed to parse value as unsigned integer".to_string())?
         } else {
             let num_mb = usize::from_str(s)
                 .map_err(|_| "failed to parse value as unsigned integer".to_string())?;
@@ -606,7 +619,7 @@ pub async fn command(config: Config) -> Result<()> {
     let frontend_shutdown = CancellationToken::new();
     let shutdown_manager = ShutdownManager::new(frontend_shutdown.clone());
 
-    let time_provider = Arc::new(SystemProvider::new());
+    let time_provider: Arc<dyn TimeProvider> = Arc::new(SystemProvider::new());
     let sys_events_store = Arc::new(SysEventStore::new(Arc::clone(&time_provider) as _));
     // setup base object store:
     let object_store: Arc<dyn ObjectStore> = config
@@ -724,7 +737,7 @@ pub async fn command(config: Config) -> Result<()> {
     let catalog = Catalog::new_with_shutdown(
         config.node_identifier_prefix.as_str(),
         Arc::clone(&object_store),
-        Arc::<SystemProvider>::clone(&time_provider),
+        Arc::clone(&time_provider),
         Arc::clone(&metrics),
         shutdown_manager.register(),
         Arc::clone(&process_uuid_getter),
@@ -798,7 +811,7 @@ pub async fn command(config: Config) -> Result<()> {
         catalog: Arc::clone(&catalog),
         last_cache,
         distinct_cache,
-        time_provider: Arc::<SystemProvider>::clone(&time_provider),
+        time_provider: Arc::clone(&time_provider),
         executor: Arc::clone(&write_path_executor),
         wal_config,
         parquet_cache,
@@ -812,6 +825,20 @@ pub async fn command(config: Config) -> Result<()> {
     .await
     .map_err(|e| Error::WriteBufferInit(e.into()))?;
 
+    let persisted_files = write_buffer_impl.persisted_files();
+
+    let object_deleter = Some(Arc::clone(&persisted_files) as _);
+
+    deleter::run(
+        DeleteManagerArgs {
+            catalog: Arc::clone(&catalog),
+            time_provider: Arc::clone(&time_provider),
+            object_deleter,
+            delete_grace_period: *config.delete_grace_period,
+        },
+        shutdown_manager.register(),
+    );
+
     info!("setting up background mem check for query buffer");
     background_buffer_checker(
         config.force_snapshot_mem_threshold.as_num_bytes(),
@@ -824,7 +851,7 @@ pub async fn command(config: Config) -> Result<()> {
         object_store_config: &config.object_store_config,
         instance_id: node_def.instance_id(),
         num_cpus,
-        persisted_files: Some(Arc::clone(&write_buffer_impl.persisted_files())),
+        persisted_files: Some(persisted_files),
         telemetry_endpoint: &config.telemetry_endpoint,
         disable_upload: config.disable_telemetry_upload,
         catalog_uuid: catalog.catalog_uuid().to_string(),
@@ -1121,6 +1148,7 @@ async fn background_buffer_checker(
 #[global_allocator]
 static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
+use influxdb3_write::deleter::DeleteManagerArgs;
 #[cfg(tokio_unstable)]
 use tokio_metrics_bridge::setup_tokio_metrics;
 

influxdb3/src/help/serve_all.txt

@@ -46,6 +46,9 @@ Examples:
                                    The amount of time that the server looks back on startup
                                    when populating the in-memory index of gen1 files.
                                   [env: INFLUXDB3_GEN1_LOOKBACK_DURATION=]
+  --delete-grace-period <DURATION> Grace period for hard deleted databases and tables before they are
+                                   removed permanently from the catalog [default: 24h]
+                                   [env: INFLUXDB3_DELETE_GRACE_PERIOD=]
 
 {}
   --aws-access-key-id <KEY>        S3 access key ID [env: AWS_ACCESS_KEY_ID=] [default: ]

influxdb3/tests/server/configure.rs

@@ -1311,6 +1311,120 @@ async fn api_v3_configure_table_delete_missing_query_param() {
     assert_eq!(StatusCode::BAD_REQUEST, resp.status());
 }
 
+#[tokio::test]
+async fn api_v3_configure_table_delete_with_hard_delete_at_never() {
+    let db_name = "foo";
+    let tbl_name = "tbl";
+    let server = TestServer::spawn().await;
+    let client = server.http_client();
+    let url = format!(
+        "{base}/api/v3/configure/table?db={db_name}&table={tbl_name}&hard_delete_at=never",
+        base = server.client_addr()
+    );
+
+    server
+        .write_lp_to_db(
+            db_name,
+            format!("{tbl_name},t1=a,t2=b,t3=c f1=true,f2=\"hello\",f3=4i,f4=4u,f5=5 1000"),
+            influxdb3_client::Precision::Second,
+        )
+        .await
+        .expect("write to db");
+
+    let resp = client
+        .delete(&url)
+        .send()
+        .await
+        .expect("delete table call succeed");
+    assert_eq!(200, resp.status());
+}
+
+#[tokio::test]
+async fn api_v3_configure_table_delete_with_hard_delete_at_timestamp() {
+    let db_name = "foo";
+    let tbl_name = "tbl";
+    let server = TestServer::spawn().await;
+    let client = server.http_client();
+    let url = format!(
+        "{base}/api/v3/configure/table?db={db_name}&table={tbl_name}&hard_delete_at=2025-12-31T23:59:59Z",
+        base = server.client_addr()
+    );
+
+    server
+        .write_lp_to_db(
+            db_name,
+            format!("{tbl_name},t1=a,t2=b,t3=c f1=true,f2=\"hello\",f3=4i,f4=4u,f5=5 1000"),
+            influxdb3_client::Precision::Second,
+        )
+        .await
+        .expect("write to db");
+
+    let resp = client
+        .delete(&url)
+        .send()
+        .await
+        .expect("delete table call succeed");
+
+    assert_eq!(200, resp.status());
+}
+
+#[tokio::test]
+async fn api_v3_configure_table_delete_with_invalid_timestamp() {
+    let db_name = "foo";
+    let tbl_name = "tbl";
+    let server = TestServer::spawn().await;
+    let client = server.http_client();
+    let url = format!(
+        "{base}/api/v3/configure/table?db={db_name}&table={tbl_name}&hard_delete_at=invalid-timestamp",
+        base = server.client_addr()
+    );
+
+    server
+        .write_lp_to_db(
+            db_name,
+            format!("{tbl_name},t1=a,t2=b,t3=c f1=true,f2=\"hello\",f3=4i,f4=4u,f5=5 1000"),
+            influxdb3_client::Precision::Second,
+        )
+        .await
+        .expect("write to db");
+
+    let resp = client
+        .delete(&url)
+        .send()
+        .await
+        .expect("delete table call succeed");
+    assert_eq!(StatusCode::BAD_REQUEST, resp.status());
+}
+
+#[tokio::test]
+async fn api_v3_configure_table_delete_with_past_timestamp_becomes_now() {
+    let db_name = "foo";
+    let tbl_name = "tbl";
+    let server = TestServer::spawn().await;
+    let client = server.http_client();
+    let url = format!(
+        "{base}/api/v3/configure/table?db={db_name}&table={tbl_name}&hard_delete_at=2020-01-01T00:00:00Z",
+        base = server.client_addr()
+    );
+
+    server
+        .write_lp_to_db(
+            db_name,
+            format!("{tbl_name},t1=a,t2=b,t3=c f1=true,f2=\"hello\",f3=4i,f4=4u,f5=5 1000"),
+            influxdb3_client::Precision::Second,
+        )
+        .await
+        .expect("write to db");
+
+    let resp = client
+        .delete(&url)
+        .send()
+        .await
+        .expect("delete table call succeed");
+
+    assert_eq!(200, resp.status());
+}
+
 #[tokio::test]
 async fn try_deleting_table_after_db_is_deleted() {
     let db_name = "db";

influxdb3_cache/src/distinct_cache/provider.rs

@@ -1,6 +1,11 @@
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
+use super::{
+    CacheError,
+    cache::{CreateDistinctCacheArgs, DistinctCache},
+};
 use arrow::datatypes::SchemaRef;
+use influxdb3_catalog::catalog::IfNotDeleted;
 use influxdb3_catalog::{
     catalog::Catalog,
     channel::CatalogUpdateReceiver,
@@ -14,11 +19,6 @@ use influxdb3_wal::{WalContents, WalOp};
 use iox_time::TimeProvider;
 use parking_lot::RwLock;
 
-use super::{
-    CacheError,
-    cache::{CreateDistinctCacheArgs, DistinctCache},
-};
-
 #[derive(Debug, thiserror::Error)]
 pub enum ProviderError {
     #[error("cache error: {0}")]
@@ -54,8 +54,12 @@ impl DistinctCacheProvider {
             catalog: Arc::clone(&catalog),
             cache_map: Default::default(),
         });
-        for db_schema in catalog.list_db_schema() {
-            for table_def in db_schema.tables() {
+        for db_schema in catalog
+            .list_db_schema()
+            .into_iter()
+            .filter_map(IfNotDeleted::if_not_deleted)
+        {
+            for table_def in db_schema.tables().filter_map(IfNotDeleted::if_not_deleted) {
                 for (cache_id, cache_def) in table_def.distinct_caches.iter() {
                     provider.create_cache(
                         db_schema.id,