diff --git a/src/log-query/src/log_query.rs b/src/log-query/src/log_query.rs index 2955b7bded..78e622200d 100644 --- a/src/log-query/src/log_query.rs +++ b/src/log-query/src/log_query.rs @@ -214,9 +214,9 @@ impl TimeFilter { (Some(start), Some(end), _) => { // Both 'start' and 'end' are provided let (start, _) = Self::parse_datetime(start)?; - let (end, _) = Self::parse_datetime(end)?; + let (end, inferred_end) = Self::parse_datetime(end)?; start_dt = Some(start); - end_dt = Some(end); + end_dt = Some(inferred_end.unwrap_or(end)); } (Some(start), None, Some(span)) => { let (start, _) = Self::parse_datetime(start)?; @@ -276,41 +276,39 @@ impl TimeFilter { if let Ok(dt) = DateTime::parse_from_rfc3339(s) { Ok((dt.with_timezone(&Utc), None)) } else { - let formats = ["%Y-%m-%d", "%Y-%m", "%Y"]; - for format in &formats { - if let Ok(naive_date) = NaiveDate::parse_from_str(s, format) { - let start = Utc.from_utc_datetime( - &naive_date.and_time(NaiveTime::from_hms_opt(0, 0, 0).unwrap()), - ); - let end = match *format { - "%Y-%m-%d" => start + Duration::days(1), - "%Y-%m" => { - let next_month = if naive_date.month() == 12 { - NaiveDate::from_ymd_opt(naive_date.year() + 1, 1, 1).unwrap() - } else { - NaiveDate::from_ymd_opt( - naive_date.year(), - naive_date.month() + 1, - 1, - ) - .unwrap() - }; - Utc.from_utc_datetime(&next_month.and_hms_opt(0, 0, 0).unwrap()) - } - "%Y" => { - let next_year = - NaiveDate::from_ymd_opt(naive_date.year() + 1, 1, 1).unwrap(); - Utc.from_utc_datetime(&next_year.and_hms_opt(0, 0, 0).unwrap()) - } - _ => unreachable!(), - }; - return Ok((start, Some(end))); + let (naive_date, format) = if let Ok(date) = NaiveDate::parse_from_str(s, "%Y-%m-%d") { + (date, "%Y-%m-%d") + } else if let Ok(date) = NaiveDate::parse_from_str(&format!("{s}-01"), "%Y-%m-%d") { + (date, "%Y-%m") + } else if let Ok(date) = NaiveDate::parse_from_str(&format!("{s}-01-01"), "%Y-%m-%d") { + (date, "%Y") + } else { + return Err(InvalidDateFormatSnafu { + input: s.to_string(), } - } - Err(InvalidDateFormatSnafu { - input: s.to_string(), - } - .build()) + .build()); + }; + + let start = Utc + .from_utc_datetime(&naive_date.and_time(NaiveTime::from_hms_opt(0, 0, 0).unwrap())); + let end = match format { + "%Y-%m-%d" => start + Duration::days(1), + "%Y-%m" => { + let next_month = if naive_date.month() == 12 { + NaiveDate::from_ymd_opt(naive_date.year() + 1, 1, 1).unwrap() + } else { + NaiveDate::from_ymd_opt(naive_date.year(), naive_date.month() + 1, 1) + .unwrap() + }; + Utc.from_utc_datetime(&next_month.and_hms_opt(0, 0, 0).unwrap()) + } + "%Y" => { + let next_year = NaiveDate::from_ymd_opt(naive_date.year() + 1, 1, 1).unwrap(); + Utc.from_utc_datetime(&next_year.and_hms_opt(0, 0, 0).unwrap()) + } + _ => unreachable!(), + }; + Ok((start, Some(end))) } } @@ -545,4 +543,44 @@ mod tests { let result = tf.canonicalize(); assert!(matches!(result, Err(Error::EndBeforeStart { .. }))); } + + #[test] + fn test_canonicalize_date_end() { + let mut date_end = TimeFilter { + start: Some("2023-10-01".to_string()), + end: Some("2023-10-02".to_string()), + span: None, + }; + date_end.canonicalize().unwrap(); + assert_eq!(date_end.end.as_deref(), Some("2023-10-03T00:00:00+00:00")); + + let mut timestamp_end = TimeFilter { + start: Some("2023-10-01".to_string()), + end: Some("2023-10-02T12:34:56Z".to_string()), + span: None, + }; + timestamp_end.canonicalize().unwrap(); + assert_eq!( + timestamp_end.end.as_deref(), + Some("2023-10-02T12:34:56+00:00") + ); + + let mut month = TimeFilter { + start: Some("2023-10".to_string()), + end: Some("2023-10".to_string()), + span: None, + }; + month.canonicalize().unwrap(); + assert_eq!(month.start.as_deref(), Some("2023-10-01T00:00:00+00:00")); + assert_eq!(month.end.as_deref(), Some("2023-11-01T00:00:00+00:00")); + + let mut year = TimeFilter { + start: Some("2023".to_string()), + end: Some("2023".to_string()), + span: None, + }; + year.canonicalize().unwrap(); + assert_eq!(year.start.as_deref(), Some("2023-01-01T00:00:00+00:00")); + assert_eq!(year.end.as_deref(), Some("2024-01-01T00:00:00+00:00")); + } } diff --git a/src/query/src/log_query/planner.rs b/src/query/src/log_query/planner.rs index f9c641b23b..53d49f98db 100644 --- a/src/query/src/log_query/planner.rs +++ b/src/query/src/log_query/planner.rs @@ -126,15 +126,14 @@ impl LogQueryPlanner { .clone(); let start_time = ScalarValue::Utf8(time_filter.start.clone()); - let end_time = ScalarValue::Utf8( - time_filter - .end - .clone() - .or(Some("9999-12-31T23:59:59Z".to_string())), - ); let expr = col(timestamp_col.clone()) .gt_eq(lit(start_time)) - .and(col(timestamp_col).lt_eq(lit(end_time))); + .and(match &time_filter.end { + Some(end) => col(timestamp_col).lt(lit(ScalarValue::Utf8(Some(end.clone())))), + None => col(timestamp_col).lt_eq(lit(ScalarValue::Utf8(Some( + "9999-12-31T23:59:59Z".to_string(), + )))), + }); Ok(expr) } @@ -730,7 +729,7 @@ mod tests { let plan = planner.query_to_plan(log_query).await.unwrap(); let expected = "Limit: skip=0, fetch=100 [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ -\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp <= Utf8(\"2021-01-02T00:00:00Z\") AND greptime.public.test_table.message LIKE Utf8(\"%error%\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ +\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp < Utf8(\"2021-01-02T00:00:00Z\") AND greptime.public.test_table.message LIKE Utf8(\"%error%\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ \n TableScan: greptime.public.test_table [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]"; assert_eq!(plan.display_indent_schema().to_string(), expected); @@ -757,7 +756,7 @@ mod tests { .gt_eq(lit(ScalarValue::Utf8(Some( "2021-01-01T00:00:00Z".to_string(), )))) - .and(col("timestamp").lt_eq(lit(ScalarValue::Utf8(Some( + .and(col("timestamp").lt(lit(ScalarValue::Utf8(Some( "2021-01-02T00:00:00Z".to_string(), ))))); @@ -851,7 +850,7 @@ mod tests { let plan = planner.query_to_plan(log_query).await.unwrap(); let expected = "Limit: skip=10, fetch=1000 [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ -\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp <= Utf8(\"2021-01-02T00:00:00Z\") AND greptime.public.test_table.message LIKE Utf8(\"%error%\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ +\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp < Utf8(\"2021-01-02T00:00:00Z\") AND greptime.public.test_table.message LIKE Utf8(\"%error%\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ \n TableScan: greptime.public.test_table [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]"; assert_eq!(plan.display_indent_schema().to_string(), expected); @@ -886,7 +885,7 @@ mod tests { let plan = planner.query_to_plan(log_query).await.unwrap(); let expected = "Limit: skip=0, fetch=1000 [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ -\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp <= Utf8(\"2021-01-02T00:00:00Z\") AND greptime.public.test_table.message LIKE Utf8(\"%error%\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ +\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp < Utf8(\"2021-01-02T00:00:00Z\") AND greptime.public.test_table.message LIKE Utf8(\"%error%\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ \n TableScan: greptime.public.test_table [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]"; assert_eq!(plan.display_indent_schema().to_string(), expected); @@ -934,7 +933,7 @@ mod tests { let plan = planner.query_to_plan(log_query).await.unwrap(); let expected = "Aggregate: groupBy=[[greptime.public.test_table.host]], aggr=[[count(greptime.public.test_table.message) AS count_result]] [host:Utf8;N, count_result:Int64]\ \n Limit: skip=0, fetch=100 [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ -\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp <= Utf8(\"2021-01-02T00:00:00Z\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ +\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp < Utf8(\"2021-01-02T00:00:00Z\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ \n TableScan: greptime.public.test_table [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]"; assert_eq!(plan.display_indent_schema().to_string(), expected); @@ -974,7 +973,7 @@ mod tests { let plan = planner.query_to_plan(log_query).await.unwrap(); let expected = "Projection: date_trunc(Utf8(\"day\"), greptime.public.test_table.timestamp) AS time_bucket [time_bucket:Timestamp(ms)]\ \n Limit: skip=0, fetch=100 [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ - \n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp <= Utf8(\"2021-01-02T00:00:00Z\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ + \n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp < Utf8(\"2021-01-02T00:00:00Z\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ \n TableScan: greptime.public.test_table [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]"; assert_eq!(plan.display_indent_schema().to_string(), expected); @@ -1058,7 +1057,7 @@ mod tests { let expected = "Aggregate: groupBy=[[2__date_histogram__time_bucket]], aggr=[[count(2__date_histogram__time_bucket) AS count_result]] [2__date_histogram__time_bucket:Timestamp(ns);N, count_result:Int64]\ \n Projection: date_bin(Utf8(\"30 seconds\"), greptime.public.test_table.timestamp) AS 2__date_histogram__time_bucket [2__date_histogram__time_bucket:Timestamp(ns);N]\ \n Limit: skip=0, fetch=1000 [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ -\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp <= Utf8(\"2021-01-02T00:00:00Z\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ +\n Filter: greptime.public.test_table.timestamp >= Utf8(\"2021-01-01T00:00:00Z\") AND greptime.public.test_table.timestamp < Utf8(\"2021-01-02T00:00:00Z\") [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]\ \n TableScan: greptime.public.test_table [message:Utf8, timestamp:Timestamp(ms), host:Utf8;N, is_active:Boolean;N]"; assert_eq!(plan.display_indent_schema().to_string(), expected); diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index c6366f48ef..a541b82b09 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -7810,14 +7810,16 @@ pub async fn test_log_query(store_type: StorageType) { .await; assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await); let res = client - .post("/v1/sql?sql=insert into logs values ('2024-11-07 10:53:50', 'hello');") + .post( + "/v1/sql?sql=insert into logs values ('2024-11-06 23:59:59', 'before-date-end'), ('2024-11-07 10:53:50', 'before-explicit-end'), ('2024-11-07 10:53:51', 'at-explicit-end'), ('2024-11-07 10:53:52', 'after-explicit-end');", + ) .header("Content-Type", "application/x-www-form-urlencoded") .send() .await; assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await); // test log query - let log_query = LogQuery { + let mut log_query = LogQuery { table: TableName { catalog_name: "greptime".to_string(), schema_name: "public".to_string(), @@ -7825,12 +7827,12 @@ pub async fn test_log_query(store_type: StorageType) { }, time_filter: TimeFilter { start: Some("2024-11-07".to_string()), - end: None, + end: Some("2024-11-07T10:53:51Z".to_string()), span: None, }, limit: Limit { skip: None, - fetch: Some(1), + fetch: Some(3), }, columns: vec!["ts".to_string(), "message".to_string()], filters: Default::default(), @@ -7847,7 +7849,36 @@ pub async fn test_log_query(store_type: StorageType) { assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await); let resp = res.text().await; let v = get_rows_from_output(&resp); - assert_eq!(v, "[[1730976830000,\"hello\"]]"); + assert_eq!(v, "[[1730976830000,\"before-explicit-end\"]]"); + + log_query.time_filter.start = Some("2024-11-06".to_string()); + log_query.time_filter.end = Some("2024-11-07".to_string()); + log_query.limit.fetch = Some(4); + let res = client + .post("/v1/logs") + .header("Content-Type", "application/json") + .body(serde_json::to_string(&log_query).unwrap()) + .send() + .await; + + assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await); + let resp = res.text().await; + let output = serde_json::from_str::(&resp).unwrap(); + let rows = output["output"][0]["records"]["rows"].as_array().unwrap(); + let mut messages = rows + .iter() + .map(|row| row[1].as_str().unwrap()) + .collect::>(); + messages.sort_unstable(); + assert_eq!( + messages, + vec![ + "after-explicit-end", + "at-explicit-end", + "before-date-end", + "before-explicit-end", + ] + ); guard.remove_all().await; }