rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-08-18 03:58:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: honor exclusive end in log queries (#8495)

Browse Source 
* fix: exclude logs end boundary

Signed-off-by: discord9 <discord9@163.com>

* fix: normalize log query date bounds

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-07-14 14:19:17 +08:00
committed by GitHub
parent 5d9dddffa7
commit 99235e4b2d
3 changed files with 123 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/log-query/src/log_query.rs
+74 -36
View File
@@ -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"));
}
}
src/query/src/log_query/planner.rs
+13 -14
View File
@@ -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);
tests-integration/tests/http.rs
+36 -5
View File
@@ -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::<Value>(&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::<Vec<_>>();
messages.sort_unstable();
assert_eq!(
messages,
vec![
"after-explicit-end",
"at-explicit-end",
"before-date-end",
"before-explicit-end",
]
);
guard.remove_all().await;
}