feat: ttl=0/instant/forever/humantime&ttl refactor

src/common/base/Cargo.toml

@@ -15,6 +15,8 @@ bytes.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 futures.workspace = true
+humantime.workspace = true
+humantime-serde.workspace = true
 paste = "1.0"
 pin-project.workspace = true
 serde = { version = "1.0", features = ["derive"] }
@@ -24,4 +26,5 @@ zeroize = { version = "1.6", default-features = false, features = ["alloc"] }
 
 [dev-dependencies]
 common-test-util.workspace = true
+serde_json.workspace = true
 toml.workspace = true

src/common/base/src/lib.rs

@@ -19,8 +19,10 @@ pub mod range_read;
 #[allow(clippy::all)]
 pub mod readable_size;
 pub mod secrets;
+pub mod ttl;
 
 pub type AffectedRows = usize;
 
 pub use bit_vec::BitVec;
 pub use plugins::Plugins;
+pub use ttl::TimeToLive;

src/common/base/src/ttl.rs

@@ -0,0 +1,181 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::Display;
+use std::time::Duration;
+
+use serde::de::Visitor;
+use serde::{Deserialize, Serialize};
+
+/// Time To Live
+
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Default)]
+pub enum TimeToLive {
+    /// immediately throw away on insert
+    Immediate,
+    /// Duration to keep the data, this duration should be non-zero
+    Duration(Duration),
+    /// Keep the data forever
+    #[default]
+    Forever,
+    // TODO(discord9): add a new variant
+    // that can't be overridden by database level ttl? call it ForceForever?
+}
+
+impl Serialize for TimeToLive {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        match self {
+            Self::Immediate => serializer.serialize_str("immediate"),
+            Self::Duration(d) => humantime_serde::serialize(d, serializer),
+            Self::Forever => serializer.serialize_str("forever"),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for TimeToLive {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct StrVisitor;
+        impl Visitor<'_> for StrVisitor {
+            type Value = TimeToLive;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("a string of time, 'immediate', 'forever' or null")
+            }
+
+            fn visit_unit<E>(self) -> Result<Self::Value, E> {
+                Ok(TimeToLive::Forever)
+            }
+
+            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                TimeToLive::from_humantime_or_str(value).map_err(serde::de::Error::custom)
+            }
+        }
+        // deser a string or null
+        let any = deserializer.deserialize_any(StrVisitor)?;
+        Ok(any)
+    }
+}
+
+impl Display for TimeToLive {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            TimeToLive::Immediate => write!(f, "immediate"),
+            TimeToLive::Duration(d) => write!(f, "Duration({})", d.as_secs()),
+            TimeToLive::Forever => write!(f, "forever"),
+        }
+    }
+}
+
+impl TimeToLive {
+    /// Parse a string into TimeToLive
+    pub fn from_humantime_or_str(s: &str) -> Result<Self, String> {
+        match s {
+            "immediate" => Ok(TimeToLive::Immediate),
+            "forever" | "" => Ok(TimeToLive::Forever),
+            _ => {
+                let d = humantime::parse_duration(s).map_err(|e| e.to_string())?;
+                Ok(TimeToLive::Duration(d))
+            }
+        }
+    }
+
+    /// Print TimeToLive as string
+    ///
+    /// omit forever variant
+    pub fn as_repr_opt(&self) -> Option<String> {
+        match self {
+            TimeToLive::Immediate => Some("immediate".to_string()),
+            TimeToLive::Duration(d) => Some(humantime::format_duration(*d).to_string()),
+            TimeToLive::Forever => None,
+        }
+    }
+
+    pub fn is_immediate(&self) -> bool {
+        matches!(self, TimeToLive::Immediate)
+    }
+
+    pub fn is_forever(&self) -> bool {
+        matches!(self, TimeToLive::Forever)
+    }
+
+    pub fn get_duration(&self) -> Option<Duration> {
+        match self {
+            TimeToLive::Duration(d) => Some(*d),
+            _ => None,
+        }
+    }
+}
+
+impl From<Duration> for TimeToLive {
+    fn from(duration: Duration) -> Self {
+        if duration.is_zero() {
+            // compatibility with old code, and inline with cassandra's behavior when ttl set to 0
+            TimeToLive::Forever
+        } else {
+            TimeToLive::Duration(duration)
+        }
+    }
+}
+
+impl From<Option<Duration>> for TimeToLive {
+    fn from(duration: Option<Duration>) -> Self {
+        match duration {
+            Some(d) => TimeToLive::from(d),
+            None => TimeToLive::Forever,
+        }
+    }
+}
+
+impl From<humantime::Duration> for TimeToLive {
+    fn from(duration: humantime::Duration) -> Self {
+        Self::from(*duration)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_serde() {
+        let cases = vec![
+            ("\"immediate\"", TimeToLive::Immediate),
+            ("\"forever\"", TimeToLive::Forever),
+            ("\"10d\"", Duration::from_secs(86400 * 10).into()),
+            (
+                "\"10000 years\"",
+                humantime::parse_duration("10000 years").unwrap().into(),
+            ),
+            ("null", TimeToLive::Forever),
+        ];
+
+        for (s, expected) in cases {
+            let serialized = serde_json::to_string(&expected).unwrap();
+            let deserialized: TimeToLive = serde_json::from_str(&serialized).unwrap();
+            assert_eq!(deserialized, expected);
+
+            let deserialized: TimeToLive = serde_json::from_str(s).unwrap();
+            assert_eq!(deserialized, expected);
+        }
+    }
+}

src/common/meta/src/ddl/alter_database.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use async_trait::async_trait;
+use common_base::TimeToLive;
 use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
 use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
 use common_telemetry::tracing::info;
@@ -46,19 +47,15 @@ fn build_new_schema_value(
             for option in options.0.iter() {
                 match option {
                     SetDatabaseOption::Ttl(ttl) => {
-                        if ttl.is_zero() {
-                            value.ttl = None;
-                        } else {
-                            value.ttl = Some(*ttl);
-                        }
+                        value.ttl = *ttl;
                     }
                 }
             }
         }
         AlterDatabaseKind::UnsetDatabaseOptions(keys) => {
             for key in keys.0.iter() {
                 match key {
-                    UnsetDatabaseOption::Ttl => value.ttl = None,
+                    UnsetDatabaseOption::Ttl => value.ttl = TimeToLive::default(),
                 }
             }
         }
@@ -220,6 +217,8 @@ impl AlterDatabaseData {
 mod tests {
     use std::time::Duration;
 
+    use common_base::TimeToLive;
+
     use crate::ddl::alter_database::build_new_schema_value;
     use crate::key::schema_name::SchemaNameValue;
     use crate::rpc::ddl::{
@@ -230,19 +229,19 @@ mod tests {
     #[test]
     fn test_build_new_schema_value() {
         let set_ttl = AlterDatabaseKind::SetDatabaseOptions(SetDatabaseOptions(vec![
-            SetDatabaseOption::Ttl(Duration::from_secs(10)),
+            SetDatabaseOption::Ttl(Duration::from_secs(10).into()),
         ]));
         let current_schema_value = SchemaNameValue::default();
         let new_schema_value =
             build_new_schema_value(current_schema_value.clone(), &set_ttl).unwrap();
-        assert_eq!(new_schema_value.ttl, Some(Duration::from_secs(10)));
+        assert_eq!(new_schema_value.ttl, Duration::from_secs(10).into());
 
         let unset_ttl_alter_kind =
             AlterDatabaseKind::UnsetDatabaseOptions(UnsetDatabaseOptions(vec![
                 UnsetDatabaseOption::Ttl,
             ]));
         let new_schema_value =
             build_new_schema_value(current_schema_value, &unset_ttl_alter_kind).unwrap();
-        assert_eq!(new_schema_value.ttl, None);
+        assert_eq!(new_schema_value.ttl, TimeToLive::default());
     }
 }