feat: warning when query's timestamp is exceeding table's ttl

Cargo.toml

@@ -87,10 +87,10 @@ lazy_static = "1.4.0"
 log = "0.4"
 logger = { path = "components/logger" }
 lru = "0.7.6"
-influxql-logical-planner = { git = "https://github.com/CeresDB/influxql", rev = "935e037a5ad6eb142a93f3e9eb321ee72e28cbad", package = "iox_query_influxql" }
-influxql-parser = { git = "https://github.com/CeresDB/influxql", rev = "935e037a5ad6eb142a93f3e9eb321ee72e28cbad", package = "influxdb_influxql_parser" }
-influxql-query = { git = "https://github.com/CeresDB/influxql", rev = "935e037a5ad6eb142a93f3e9eb321ee72e28cbad", package = "iox_query" }
-influxql-schema = { git = "https://github.com/CeresDB/influxql", rev = "935e037a5ad6eb142a93f3e9eb321ee72e28cbad", package = "schema" }
+influxql-logical-planner = { git = "https://github.com/CeresDB/influxql", rev = "b65a125b9cdfa3121a3c8843bc48441b91049e31", package = "iox_query_influxql" }
+influxql-parser = { git = "https://github.com/CeresDB/influxql", rev = "b65a125b9cdfa3121a3c8843bc48441b91049e31", package = "influxdb_influxql_parser" }
+influxql-query = { git = "https://github.com/CeresDB/influxql", rev = "b65a125b9cdfa3121a3c8843bc48441b91049e31", package = "iox_query" }
+influxql-schema = { git = "https://github.com/CeresDB/influxql", rev = "b65a125b9cdfa3121a3c8843bc48441b91049e31", package = "schema" }
 interpreters = { path = "interpreters" }
 itertools = "0.10.5"
 meta_client = { path = "meta_client" }

analytic_engine/src/compaction/mod.rs

@@ -4,6 +4,7 @@
 
 use std::{collections::HashMap, fmt, str::FromStr, sync::Arc};
 
+use common_types::COMPACTION_STRATEGY;
 use common_util::config::{ReadableSize, TimeUnit};
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, Backtrace, GenerateBacktrace, ResultExt, Snafu};
@@ -13,7 +14,6 @@ use crate::{
     compaction::picker::{CommonCompactionPicker, CompactionPickerRef},
     sst::file::{FileHandle, Level},
     table::data::TableDataRef,
-    table_options::COMPACTION_STRATEGY,
 };
 
 mod metrics;

analytic_engine/src/table_options.rs

@@ -5,32 +5,24 @@
 use std::{collections::HashMap, string::ToString, time::Duration};
 
 use ceresdbproto::manifest as manifest_pb;
-use common_types::time::Timestamp;
+use common_types::{
+    time::Timestamp, ARENA_BLOCK_SIZE, COMPACTION_STRATEGY, COMPRESSION, ENABLE_TTL,
+    NUM_ROWS_PER_ROW_GROUP, OPTION_KEY_ENABLE_TTL, SEGMENT_DURATION, STORAGE_FORMAT, TTL,
+    UPDATE_MODE, WRITE_BUFFER_SIZE,
+};
 use common_util::{
     config::{ReadableDuration, ReadableSize, TimeUnit},
     define_result,
-    time::DurationExt,
+    time::{parse_duration, DurationExt},
 };
 use datafusion::parquet::basic::Compression as ParquetCompression;
 use serde::{Deserialize, Serialize};
 use snafu::{Backtrace, GenerateBacktrace, OptionExt, ResultExt, Snafu};
-use table_engine::OPTION_KEY_ENABLE_TTL;
 
 use crate::compaction::{
     self, CompactionStrategy, SizeTieredCompactionOptions, TimeWindowCompactionOptions,
 };
 
-pub const SEGMENT_DURATION: &str = "segment_duration";
-pub const ENABLE_TTL: &str = OPTION_KEY_ENABLE_TTL;
-pub const TTL: &str = "ttl";
-pub const ARENA_BLOCK_SIZE: &str = "arena_block_size";
-pub const WRITE_BUFFER_SIZE: &str = "write_buffer_size";
-pub const COMPACTION_STRATEGY: &str = "compaction_strategy";
-pub const NUM_ROWS_PER_ROW_GROUP: &str = "num_rows_per_row_group";
-pub const UPDATE_MODE: &str = "update_mode";
-pub const COMPRESSION: &str = "compression";
-pub const STORAGE_FORMAT: &str = "storage_format";
-
 const UPDATE_MODE_OVERWRITE: &str = "OVERWRITE";
 const UPDATE_MODE_APPEND: &str = "APPEND";
 const COMPRESSION_UNCOMPRESSED: &str = "UNCOMPRESSED";
@@ -64,8 +56,11 @@ const MAX_NUM_ROWS_PER_ROW_GROUP: usize = 10_000_000;
 #[derive(Debug, Snafu)]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
-    #[snafu(display("Failed to parse duration, err:{}.\nBacktrace:\n{}", err, backtrace))]
-    ParseDuration { err: String, backtrace: Backtrace },
+    #[snafu(display("Failed to parse duration, err:{}.\nBacktrace:\n{}", source, backtrace))]
+    ParseDuration {
+        source: common_util::time::Error,
+        backtrace: Backtrace,
+    },
 
     #[snafu(display("Failed to parse size, err:{}.\nBacktrace:\n{}", err, backtrace))]
     ParseSize { err: String, backtrace: Backtrace },
@@ -707,15 +702,15 @@ fn merge_table_options(
     let mut table_opts = table_old_opts.clone();
     if is_create {
         if let Some(v) = options.get(SEGMENT_DURATION) {
-            table_opts.segment_duration = Some(parse_duration(v)?);
+            table_opts.segment_duration = Some(parse_duration(v).context(ParseDuration)?);
         }
         if let Some(v) = options.get(UPDATE_MODE) {
             table_opts.update_mode = UpdateMode::parse_from(v)?;
         }
     }
 
     if let Some(v) = options.get(TTL) {
-        table_opts.ttl = parse_duration(v)?;
+        table_opts.ttl = parse_duration(v).context(ParseDuration)?;
     }
     if let Some(v) = options.get(OPTION_KEY_ENABLE_TTL) {
         table_opts.enable_ttl = v.parse::<bool>().context(ParseBool)?;
@@ -744,14 +739,6 @@ fn merge_table_options(
     Ok(table_opts)
 }
 
-fn parse_duration(v: &str) -> Result<ReadableDuration> {
-    v.parse::<ReadableDuration>()
-        .map_err(|err| Error::ParseDuration {
-            err,
-            backtrace: Backtrace::generate(),
-        })
-}
-
 fn parse_size(v: &str) -> Result<ReadableSize> {
     v.parse::<ReadableSize>().map_err(|err| Error::ParseSize {
         err,

analytic_engine/src/tests/table.rs

@@ -277,7 +277,7 @@ impl Builder {
 
     pub fn enable_ttl(mut self, enable_ttl: bool) -> Self {
         self.create_request.options.insert(
-            table_engine::OPTION_KEY_ENABLE_TTL.to_string(),
+            common_types::OPTION_KEY_ENABLE_TTL.to_string(),
             enable_ttl.to_string(),
         );
         self
@@ -286,7 +286,7 @@ impl Builder {
     pub fn ttl(mut self, duration: ReadableDuration) -> Self {
         self.create_request
             .options
-            .insert(table_options::TTL.to_string(), duration.to_string());
+            .insert(common_types::TTL.to_string(), duration.to_string());
         self
     }
 

common_types/src/lib.rs

@@ -31,5 +31,18 @@ pub const MAX_SEQUENCE_NUMBER: u64 = u64::MAX;
 /// sequence number should starts from 1.
 pub const MIN_SEQUENCE_NUMBER: u64 = 0;
 
+/// Enable ttl key
+pub const OPTION_KEY_ENABLE_TTL: &str = "enable_ttl";
+pub const SEGMENT_DURATION: &str = "segment_duration";
+pub const ENABLE_TTL: &str = OPTION_KEY_ENABLE_TTL;
+pub const TTL: &str = "ttl";
+pub const ARENA_BLOCK_SIZE: &str = "arena_block_size";
+pub const WRITE_BUFFER_SIZE: &str = "write_buffer_size";
+pub const COMPACTION_STRATEGY: &str = "compaction_strategy";
+pub const NUM_ROWS_PER_ROW_GROUP: &str = "num_rows_per_row_group";
+pub const UPDATE_MODE: &str = "update_mode";
+pub const COMPRESSION: &str = "compression";
+pub const STORAGE_FORMAT: &str = "storage_format";
+
 #[cfg(any(test, feature = "test"))]
 pub mod tests;

common_util/src/time.rs

@@ -11,6 +11,18 @@ use std::{
 
 use chrono::{DateTime, Utc};
 use common_types::time::Timestamp;
+use snafu::{Backtrace, GenerateBacktrace, Snafu};
+
+use crate::config::ReadableDuration;
+
+#[derive(Debug, Snafu)]
+#[allow(clippy::enum_variant_names)]
+pub enum Error {
+    #[snafu(display("Failed to parse duration, err:{}.\nBacktrace:\n{}", err, backtrace))]
+    ParseDuration { err: String, backtrace: Backtrace },
+}
+
+define_result!(Error);
 
 pub trait DurationExt {
     /// Convert into u64.
@@ -78,6 +90,14 @@ pub fn try_to_millis(ts: i64) -> Option<Timestamp> {
     None
 }
 
+pub fn parse_duration(v: &str) -> Result<ReadableDuration> {
+    v.parse::<ReadableDuration>()
+        .map_err(|err| Error::ParseDuration {
+            err,
+            backtrace: Backtrace::generate(),
+        })
+}
+
 #[cfg(test)]
 mod tests {
     use std::thread;

proxy/Cargo.toml

@@ -26,6 +26,7 @@ df_operator = { workspace = true }
 futures = { workspace = true }
 http = "0.2"
 influxdb-line-protocol = "1.0"
+influxql-query = { workspace = true }
 interpreters = { workspace = true }
 itertools = { workspace = true }
 lazy_static = { workspace = true }

proxy/src/error.rs

@@ -27,13 +27,17 @@ pub enum Error {
         msg: String,
         source: GenericError,
     },
+
+    #[snafu(display("Query warning, msg:{msg}"))]
+    QueryMaybeExceedTTL { msg: String },
 }
 
 impl Error {
     pub fn code(&self) -> StatusCode {
         match *self {
             Error::ErrNoCause { code, .. } => code,
             Error::ErrWithCause { code, .. } => code,
+            Error::QueryMaybeExceedTTL { .. } => StatusCode::OK,
             Error::Internal { .. } | Error::InternalNoCause { .. } => {
                 StatusCode::INTERNAL_SERVER_ERROR
             }
@@ -50,6 +54,7 @@ impl Error {
                 let first_line = error_util::remove_backtrace_from_err(&err_string);
                 format!("{msg}. Caused by: {first_line}")
             }
+            Error::QueryMaybeExceedTTL { msg } => msg.clone(),
         }
     }
 }