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

analytic_engine/src/table_options.rs

@@ -744,7 +744,7 @@ fn merge_table_options(
     Ok(table_opts)
 }
 
-fn parse_duration(v: &str) -> Result<ReadableDuration> {
+pub fn parse_duration(v: &str) -> Result<ReadableDuration> {
     v.parse::<ReadableDuration>()
         .map_err(|err| Error::ParseDuration {
             err,

proxy/Cargo.toml

@@ -11,6 +11,7 @@ workspace = true
 workspace = true
 
 [dependencies]
+analytic_engine = { workspace = true }
 arrow = { workspace = true }
 arrow_ext = { workspace = true }
 async-trait = { workspace = true }
@@ -26,6 +27,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("sql query error, message:{}", msg))]
+    SqlQueryOverTTL { code: StatusCode, msg: String },
 }
 
 impl Error {
     pub fn code(&self) -> StatusCode {
         match *self {
             Error::ErrNoCause { code, .. } => code,
             Error::ErrWithCause { code, .. } => code,
+            Error::SqlQueryOverTTL { code, .. } => code,
             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::SqlQueryOverTTL { code: _, msg } => msg.clone(),
         }
     }
 }

proxy/src/lib.rs

@@ -6,7 +6,7 @@
 #![feature(trait_alias)]
 
 pub mod context;
-mod error;
+pub mod error;
 mod error_util;
 pub mod forward;
 mod grpc;
@@ -27,20 +27,26 @@ pub const FORWARDED_FROM: &str = "forwarded-from";
 
 use std::{
     sync::Arc,
-    time::{Duration, Instant},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
 use ::http::StatusCode;
-use catalog::schema::{
-    CreateOptions, CreateTableRequest, DropOptions, DropTableRequest, NameRef, SchemaRef,
+use analytic_engine::table_options::{parse_duration, ENABLE_TTL, TTL};
+use catalog::{
+    schema::{
+        CreateOptions, CreateTableRequest, DropOptions, DropTableRequest, NameRef, SchemaRef,
+    },
+    CatalogRef,
 };
 use ceresdbproto::storage::{
     storage_service_client::StorageServiceClient, PrometheusRemoteQueryRequest,
     PrometheusRemoteQueryResponse, Route, RouteRequest,
 };
 use common_types::{request_id::RequestId, table::DEFAULT_SHARD_ID};
 use common_util::{error::BoxError, runtime::Runtime};
+use datafusion::prelude::{Column, Expr};
 use futures::FutureExt;
+use influxql_query::logical_optimizer::range_predicate::find_time_range;
 use interpreters::{
     context::Context as InterpreterContext,
     factory::Factory,
@@ -54,7 +60,7 @@ use snafu::{OptionExt, ResultExt};
 use table_engine::{
     engine::{EngineRuntimes, TableState},
     remote::model::{GetTableInfoRequest, TableIdentifier},
-    table::TableId,
+    table::{TableId, TableRef},
     PARTITION_TABLE_ENGINE_TYPE,
 };
 use tonic::{transport::Channel, IntoRequest};
@@ -165,6 +171,118 @@ impl<Q: QueryExecutor + 'static> Proxy<Q> {
         })
     }
 
+    fn valid_ttl_range(
+        &self,
+        plan: &Plan,
+        catalog_name: &str,
+        schema_name: &str,
+        table_name: &str,
+    ) -> bool {
+        if let Plan::Query(query) = &plan {
+            let tableref = match self.get_table(catalog_name, schema_name, table_name) {
+                Ok(Some(tableref)) => tableref,
+                _ => return true,
+            };
+            if let Some(value) = tableref.options().get(ENABLE_TTL) {
+                if value == "false" {
+                    return true;
+                }
+            }
+
+            let ttl_duration = match tableref.options().get(TTL) {
+                Some(value) => parse_duration(value),
+                None => return false,
+            };
+            assert!(ttl_duration.is_ok(), "ttl_duration muse be vaild");
+            let ttl_duration = ttl_duration.unwrap();
+
+            // TODO(tanruixiang): use sql's timestamp as nowtime(need sql support)
+            let nowtime = match SystemTime::now().duration_since(UNIX_EPOCH) {
+                Ok(duration) => duration.as_secs() as i64,
+                Err(_) => {
+                    return false;
+                }
+            };
+
+            // Because the clock may have errors, choose 1 hour as the error buffer
+            let buffer_duration = Duration::new(60 * 60, 0);
+
+            let ddl = nowtime - ttl_duration.as_secs() as i64 - buffer_duration.as_secs() as i64;
+
+            let timestamp_name = &tableref
+                .schema()
+                .column(tableref.schema().timestamp_index())
+                .name
+                .clone();
+            let ts_col = Column::from_name(timestamp_name);
+            let range = find_time_range(&query.df_plan, &ts_col).unwrap();
+            match range.end {
+                std::ops::Bound::Included(x) | std::ops::Bound::Excluded(x) => {
+                    if let Expr::Literal(datafusion::scalar::ScalarValue::Int64(Some(x))) = x {
+                        if x <= ddl {
+                            return false;
+                        }
+                    }
+                }
+                std::ops::Bound::Unbounded => (),
+            }
+        }
+        true
+    }
+
+    fn get_catalog(&self, catalog_name: &str) -> Result<CatalogRef> {
+        let catalog = self
+            .instance
+            .catalog_manager
+            .catalog_by_name(catalog_name)
+            .box_err()
+            .with_context(|| ErrWithCause {
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+                msg: format!("Failed to find catalog, catalog_name:{catalog_name}"),
+            })?
+            .with_context(|| ErrNoCause {
+                code: StatusCode::BAD_REQUEST,
+                msg: format!("Catalog not found, catalog_name:{catalog_name}"),
+            })?;
+        Ok(catalog)
+    }
+
+    fn get_schema(&self, catalog: &CatalogRef, schema_name: &str) -> Result<SchemaRef> {
+        // TODO: support create schema if not exist
+        let schema = catalog
+            .schema_by_name(schema_name)
+            .box_err()
+            .with_context(|| ErrWithCause {
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+                msg: format!("Failed to find schema, schema_name:{schema_name}"),
+            })?
+            .context(ErrNoCause {
+                code: StatusCode::BAD_REQUEST,
+                msg: format!("Schema not found, schema_name:{schema_name}"),
+            })?;
+        Ok(schema)
+    }
+
+    fn get_table(
+        &self,
+        catalog_name: &str,
+        schema_name: &str,
+        table_name: &str,
+    ) -> Result<Option<TableRef>> {
+        let catalog = self.get_catalog(catalog_name)?;
+
+        let schema = self.get_schema(&catalog, schema_name)?;
+
+        let table = schema
+            .table_by_name(table_name)
+            .box_err()
+            .with_context(|| ErrWithCause {
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+                msg: format!("Failed to find table, table_name:{table_name}"),
+            })?;
+        Ok(table)
+    }
+
     async fn maybe_open_partition_table_if_not_exist(
         &self,
         catalog_name: &str,

proxy/src/read.rs

@@ -115,8 +115,8 @@ impl<Q: QueryExecutor + 'static> Proxy<Q> {
 
         // Open partition table if needed.
         let table_name = frontend::parse_table_name(&stmts);
-        if let Some(table_name) = table_name {
-            self.maybe_open_partition_table_if_not_exist(catalog, schema, &table_name)
+        if let Some(table_name) = &table_name {
+            self.maybe_open_partition_table_if_not_exist(catalog, schema, table_name)
                 .await?;
         }
 
@@ -132,6 +132,15 @@ impl<Q: QueryExecutor + 'static> Proxy<Q> {
                 msg: format!("Failed to create plan, query:{sql}"),
             })?;
 
+        if let Some(table_name) = &table_name {
+            if !self.valid_ttl_range(&plan, catalog, schema, table_name) {
+                return Err(Error::SqlQueryOverTTL {
+                    code: StatusCode::OK,
+                    msg: format!("The query {sql}'s timestamp is already out of date"),
+                });
+            }
+        }
+
         let output = if ctx.enable_partition_table_access {
             self.execute_plan_involving_partition_table(request_id, catalog, schema, plan, deadline)
                 .await

server/src/http.rs

@@ -115,6 +115,9 @@ pub enum Error {
 
     #[snafu(display("Missing wal.\nBacktrace:\n{}", backtrace))]
     MissingWal { backtrace: Backtrace },
+
+    #[snafu(display("sql query error, message:{}", msg))]
+    SqlQueryOverTTL { code: StatusCode, msg: String },
 }
 
 define_result!(Error);
@@ -260,12 +263,18 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
                 let result = proxy
                     .handle_http_sql_query(&ctx, req)
                     .await
-                    .map(convert_output)
-                    .box_err()
-                    .context(HandleRequest);
+                    .map(convert_output);
+
                 match result {
                     Ok(res) => Ok(reply::json(&res)),
-                    Err(e) => Err(reject::custom(e)),
+                    Err(e) => {
+                        if let proxy::error::Error::SqlQueryOverTTL { code, msg } = e {
+                            return Err(reject::custom(Error::SqlQueryOverTTL { code, msg }));
+                        }
+                        Err(reject::custom(Error::Internal {
+                            source: Box::new(e),
+                        }))
+                    }
                 }
             })
     }
@@ -723,6 +732,7 @@ fn error_to_status_code(err: &Error) -> StatusCode {
         | Error::MissingRouter { .. }
         | Error::MissingWal { .. }
         | Error::HandleUpdateLogLevel { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+        Error::SqlQueryOverTTL { code: _, msg: _ } => StatusCode::OK,
     }
 }