rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-27 22:26:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: backport suggested independent fixes to release/5e02efa9 (#9795)

Browse Source 
* fix(servers): resolve port from IPv6 bind addresses (#9142)

Signed-off-by: immanuwell <pchpr.00@list.ru>
(cherry picked from commit 5ec679fed9)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: register region migration procedures before returning IDs (#9163)

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
(cherry picked from commit c65e8d0962)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(promql): skip empty batches in SeriesDivide stream (#9178)

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
(cherry picked from commit 5e57b00a95)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(servers): merge duplicate timeseries across RecordBatches in Prometheus remote read (#9032)

* fix(servers): merge timeseries across RecordBatches in Prometheus remote read

Signed-off-by: Tyagiquamar <mohdquamartyagi@gmail.com>

* fix(servers): exercise recordbatches_to_timeseries and sort samples with sort_unstable_by_key

---------

Signed-off-by: Tyagiquamar <mohdquamartyagi@gmail.com>
Co-authored-by: dennis zhuang <killme2008@gmail.com>
(cherry picked from commit 59eb63c663)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(prometheus): honor label matchers in __name__ values query (#9134)

* fix(prometheus): honor label matchers in __name__ values query

`/api/v1/label/__name__/values?match[]={pod="abc"}` dropped every matcher
other than `__name__` and returned all metrics in the schema. No error,
just the wrong list. Grafana's metrics browser sends this request, so
picking a label value there did nothing.

Selectors that only constrain `__name__` keep answering from table
metadata. A selector constraining an ordinary label now goes to the data:
scan each metric engine physical table for distinct `__table_id` in the
time range, map the ids back to metric names, then apply the selector's
own `__name__` matchers.

Only metric engine tables are covered; other engines share no column space
to scan.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(prometheus): batch-resolve metric names by table id

Building a full table-id-to-name map meant walking every table in the
schema and holding all of them in memory, just to name the handful the
scan returned. Use `tables_by_ids` instead — one batch KV read over the
ids the scan actually produced.

The catalog walk stays, but only to find the physical tables to scan.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(promql): read an absent label as the empty string

A matcher on a label the series does not carry only worked when the table
had no column for it at all. Where the column exists but is NULL on that
row -- the norm for logical metrics sharing a metric engine physical
table, which holds the union of their label columns -- three-valued logic
dropped the row, so `host!="host1"` and `host=""` missed every metric
without a host label.

Coalesce nullable string label columns to "" for matchers that accept the
empty string, rather than only for the OTLP temporality marker. Equality
matchers are untouched; they cannot match NULL either way.

This is the Prometheus compatibility fix #8970 deliberately kept out of
its own scope. The cost is visible in the regex sqlness plan: the
predicate becomes a CASE, so the scan loses its LastRow selector and
grows a FilterExec. Only negative and empty-accepting matchers pay it.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(promql): don't panic on pre-epoch label value bounds

`rewrite_label_values_query` unwrapped `duration_since(UNIX_EPOCH)`, which
returns an error for an instant before the epoch. `start=1969-12-31T23:59:59Z`
parses as valid RFC3339, so the request panicked instead of answering.

Recover the sign from the error branch, and report a value beyond i64
milliseconds as an error rather than wrapping the cast.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(prometheus): drop applicable_matchers, share the distinct scan

With the planner reading an absent label as empty, the frontend no longer
needs to pre-filter matchers per physical table. Removing that exposed a
second problem: a physical table that never took a column from a logical
table exposes no `__table_id`, and projecting it failed the whole request.
Skip those tables; the only thing that can miss is a metric with no labels.

Also pulls out the plan-build-execute-collect sequence the two label value
scans had in common.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit 007836ba7a)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): cancel cache construction for incomplete scans (#9254)

* fix(mito2): cancel cache construction for incomplete scans

CacheBatchBuffer spawned the background concat task and dropped its join
handle. A scan that was cancelled or failed therefore left the task alive:
it kept compacting already queued batches, and while waiting for a range
result memory permit it held them, even though without a finish command
the result can never be put into the cache.

Keep the handle and abort it when the buffer is dropped. The handle is
cleared once the task owns the finish command, so a completed scan still
populates the cache after its stream is dropped. Abort does not preempt a
concat that is already running; it takes effect the next time the task is
polled.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(mito2): wait for the permit park before cancelling the buffer

An empty buffered_batches only proves the batches were enqueued, so the
cancellation test could abort a concat task that had never been polled.
Count acquisitions that find too few permits, a test-only signal, and drop
the buffer once the task has reached that wait. Nothing awaits between the
check and the parking, so an observed increment means the caller is about
to wait for permits the test holds for the rest of the case.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit 43ce07fa69)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: merge every state row in geo_path and json_encode_path (#9339)

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit cc1be82858)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: align ordered aggregate state type with the accumulator output (#9340)

* fix: align ordered aggregate state type with the accumulator output

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: reject mismatched aggregate states and keep hard-ordered aggregates unsplit

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: keep WITHIN GROUP aggregates splittable and relabel state fields by position

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit affc0a1b1d)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): stop parallel flat scan tasks once the receiver is dropped (#9368)

* fix(mito2): stop parallel flat scan tasks once the receiver is dropped

spawn_flat_scan_task ignored send errors, so after a query was cancelled the
detached task kept reading its source to EOF. Break out of the loop when the
receiver is gone.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore(mito2): log when a parallel scan task stops on a dropped receiver

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore(mito2): include the send error in the parallel scan stop log

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit 411c0f173e)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(catalog): skip backend batch_get on full cache hits (#9431)

fix(catalog): return cached batch values without an empty backend call

Count backend batch_get invocations in regression tests and cover empty input.

Signed-off-by: WenyXu <wenymedia@gmail.com>
(cherry picked from commit acebd6b35b)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: coalesce equal histogram quantile bucket boundaries (#9451)

* fix: coalesce equal histogram quantile bucket boundaries

Merge numerically equivalent bucket boundaries before monotonicity repair, preserving the allocation-free path for unique boundaries. Share coalescing with histogram_fraction and cover duplicate boundaries with unit and SQL regressions.

Fixes #9443

Signed-off-by: evenyag <realevenyag@gmail.com>

* fix: keep duplicate infinity buckets in one histogram group

Confirm classic histogram completion at a series/timestamp boundary or EOF instead of the first positive infinity. Retain lookahead across batches and reject incomplete optimistic groups so equivalent infinity labels reach numeric coalescing together.

Add stream and SQL regressions for duplicate infinity labels, including established bucket counts and EOF. Correct older fixtures to use distinct keys for distinct histogram groups.

Signed-off-by: evenyag <realevenyag@gmail.com>

* fix: avoid buffering histogram groups with invalid trailing bounds

Discover candidate bucket counts at the first group boundary and let optimistic validation fall back to safe mode for malformed bounds. Document discovery and cover trailing invalid and null labels with a focused streaming regression.

Signed-off-by: evenyag <realevenyag@gmail.com>

---------

Signed-off-by: evenyag <realevenyag@gmail.com>
(cherry picked from commit 97913b4b3a)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(otlp): coerce empty and "-" trace attribute strings to null (#9461)

* fix(otlp): coerce empty and "-" trace attribute strings to null

Instrumentations fill numeric attributes from absent headers with an empty
string (PSR-7 getHeaderLine returns ""). Coercing that into an existing
numeric column failed and rejected the whole span.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test: trim redundant row from empty-string rewrite test

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit d640ab8bc5)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(metasrv): retry failed heartbeat address updates (#9437)

* fix(metasrv): retry failed heartbeat address updates

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor(metasrv): move shared heartbeat helpers to utils

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
(cherry picked from commit 65b9a35692)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: align TWCS assignment with half-open time windows (#9464)

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
(cherry picked from commit 5126623e63)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(promql): keep vector labels when a computed scalar is on the right (#9478)

* fix(promql): keep vector labels when a computed scalar is on the right

`metric + time()` or `metric * scalar(other)` planned the join with the
scalar side's context, which has no tags, so every series collapsed into
one unlabeled series. Use the vector side's context when the right operand
is scalar-typed.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(promql): update results for labels kept with a computed scalar on the right

v * pi() and v + time() now keep the vector's labels and time index name,
as in Prometheus.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(promql): qualify columns of an unnamed operand in vector joins

A comparison filter's projection has no table name, so after the join its
time index resolved unqualified and was ambiguous when the other operand had
the same column. Also keep the empty-metric branch unchanged for scalar
left operands.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(promql): cover comparison filters joined with a scalar

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
(cherry picked from commit 6da5352c53)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(promql): keep metric name for comparison operators (#9332)

* fix(promql): keep metric name for comparison operators

`collect_metric_names` dropped the response `__name__` for every binary
operator except `and` / `or` / `unless`, so comparisons such as
`node_load1 > 5` came back without the metric name. Grafana rendered them
as `{instance="..."}` and alert templates reading `$labels.__name__` got
nothing.

Follow Prometheus' `shouldDropMetricName` / `resultMetric`: arithmetic
operators drop the name, comparisons keep the left-hand name, except when
the comparison returns a bool or is a one-to-one `on(...)` match. A
`group_left` / `group_right` modifier does not reduce the labels, so the
name is kept there.

Both the HTTP PromQL API and the gRPC gateway share
`retrieve_metric_name_and_result_type`, so one fix covers both.

Signed-off-by: Aumnertic <hznuzc@gmail.com>

* docs: document binary_keeps_metric_name

Signed-off-by: Aumnertic <hznuzc@gmail.com>

* fix(promql): handle on(__name__), scalar lhs and group_right

Signed-off-by: Aumnertic <hznuzc@gmail.com>

* chore(promql): correct result-side docs and drop unreachable String arm

Signed-off-by: Aumnertic <hznuzc@gmail.com>

* fix(promql): drop metric name when explicitly ignored

Handle ignoring(__name__) for one-to-one comparisons and cover
comparison, group_left, group_right and unless behavior.

Signed-off-by: Aumnertic <hznuzc@gmail.com>

---------

Signed-off-by: Aumnertic <hznuzc@gmail.com>
(cherry picked from commit c33a6cfa56)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(query): keep filter fetch when rewriting matches term (#9521)

* test: cover matches term limit dropped by constant rewrite

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(query): keep filter fetch when rewriting matches term

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
(cherry picked from commit 483d1e49a3)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* style: format #9437 test imports for the release toolchain

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test: regenerate array_agg result for release plan rendering

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: immanuwell <pchpr.00@list.ru>
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
Signed-off-by: Tyagiquamar <mohdquamartyagi@gmail.com>
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
Signed-off-by: WenyXu <wenymedia@gmail.com>
Signed-off-by: evenyag <realevenyag@gmail.com>
Signed-off-by: Aumnertic <hznuzc@gmail.com>
Co-authored-by: Immanuel Tikhonov <122638311+immanuwell@users.noreply.github.com>
Co-authored-by: jeremyhi <jiachun_feng@proton.me>
Co-authored-by: discord9 <discord9@163.com>
Co-authored-by: Mohd Quamar Tyagi <104281681+Tyagiquamar@users.noreply.github.com>
Co-authored-by: dennis zhuang <killme2008@gmail.com>
Co-authored-by: Weny Xu <wenymedia@gmail.com>
Co-authored-by: Yingwen <realevenyag@gmail.com>
Co-authored-by: Aum <hznuzc@gmail.com>
This commit is contained in:
Lei, HUANG
2026-10-20 09:46:46 -07:00
committed by GitHub
co-authored by Immanuel Tikhonov jeremyhi discord9 Mohd Quamar Tyagi dennis zhuang Weny Xu Yingwen Aum
parent fd3faeb7ed
commit 2b05da4c06
48 changed files with 3869 additions and 613 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/cache.rs
+20
View File
@@ -101,6 +101,10 @@ pub(crate) struct RangeResultMemoryLimiter {
semaphore: Arc<tokio::sync::Semaphore>,
permit_bytes: usize,
total_permits: usize,
/// Number of acquisitions that found too few permits and parked. Test-only
/// signal to synchronize with a caller waiting inside [Self::acquire].
#[cfg(test)]
waited_acquires: std::sync::atomic::AtomicUsize,
}
impl Default for RangeResultMemoryLimiter {
@@ -122,6 +126,8 @@ impl RangeResultMemoryLimiter {
semaphore: Arc::new(tokio::sync::Semaphore::new(total_permits)),
permit_bytes,
total_permits,
#[cfg(test)]
waited_acquires: std::sync::atomic::AtomicUsize::new(0),
}
}
@@ -130,6 +136,12 @@ impl RangeResultMemoryLimiter {
self.permit_bytes
}
#[cfg(test)]
pub(crate) fn waited_acquires(&self) -> usize {
self.waited_acquires
.load(std::sync::atomic::Ordering::Acquire)
}
#[cfg(test)]
pub(crate) fn available_permits(&self) -> usize {
self.semaphore.available_permits()
@@ -146,6 +158,14 @@ impl RangeResultMemoryLimiter {
}
.fail();
}
// Nothing awaits between this check and the parking below, so an observed
// increment means the caller is about to wait for the missing permits.
#[cfg(test)]
if self.semaphore.available_permits() < permits {
self.waited_acquires
.fetch_add(1, std::sync::atomic::Ordering::Release);
}
self.semaphore
.acquire_many(permits as u32)
.await
src/mito2/src/compaction/twcs.rs
+216 -88
View File
@@ -892,6 +892,7 @@ struct Window {
start: Timestamp,
end: Timestamp,
files: Vec<FileHandle>,
/// Start of the half-open window, in seconds.
time_window: i64,
primary_key_range: Option<(bytes::Bytes, bytes::Bytes)>,
}
@@ -942,12 +943,7 @@ fn assign_to_windows<'a>(
continue;
}
let (_, end) = f.time_range();
let time_window = end
.convert_to(TimeUnit::Second)
.unwrap()
.value()
.align_to_ceil_by_bucket(time_window_size)
.unwrap_or(i64::MIN);
let time_window = time_window_start(end, time_window_size);
match windows.entry(time_window) {
Entry::Occupied(mut e) => {
@@ -963,37 +959,41 @@ fn assign_to_windows<'a>(
windows.into_iter().collect()
}
/// Uses the same half-open windows as memtables, including timestamps in the first
/// second after a boundary. An unrepresentable window start is clamped to `i64::MIN`.
fn time_window_start(timestamp: Timestamp, time_window_size: i64) -> i64 {
// Conversion to seconds cannot overflow.
timestamp
.convert_to(TimeUnit::Second)
.and_then(|ts| ts.value().align_by_bucket(time_window_size))
.unwrap_or(i64::MIN)
}
fn time_window_intersects_range(
window_end: i64,
window_start: i64,
time_window_size: i64,
time_range: &TimestampRange,
) -> bool {
let first_window = match time_range.start() {
None => i64::MIN,
Some(start) => {
let Some(first_window) = start
.convert_to(TimeUnit::Second)
.and_then(|timestamp| timestamp.value().align_to_ceil_by_bucket(time_window_size))
else {
return false;
};
first_window
}
};
if time_range.is_empty() {
return false;
}
let first_window = time_range
.start()
.map_or(i64::MIN, |start| time_window_start(start, time_window_size));
let last_window = match time_range.end() {
None => i64::MAX,
Some(end) => {
let Some(last_window) = end
// The range end is exclusive; keep the last second that intersects it.
let Some(last_second) = end
.convert_to_ceil(TimeUnit::Second)
.and_then(|timestamp| timestamp.value().checked_sub(1))
.and_then(|timestamp| timestamp.align_to_ceil_by_bucket(time_window_size))
else {
return false;
};
last_window
time_window_start(Timestamp::new_second(last_second), time_window_size)
}
};
(first_window..=last_window).contains(&window_end)
(first_window..=last_window).contains(&window_start)
}
fn window_has_overlap(this: &Window, windows: &BTreeMap<i64, Window>) -> bool {
@@ -1011,7 +1011,7 @@ fn window_has_overlap(this: &Window, windows: &BTreeMap<i64, Window>) -> bool {
}
/// Finds the latest active writing window among all files.
/// Returns `None` when there are no files or all files are corrupted.
/// Returns `None` when there are no files.
fn find_latest_window_in_seconds<'a>(
files: impl Iterator<Item = &'a FileHandle>,
time_window_size: i64,
@@ -1027,9 +1027,7 @@ fn find_latest_window_in_seconds<'a>(
latest_timestamp = Some(end);
}
}
latest_timestamp
.and_then(|ts| ts.convert_to_ceil(TimeUnit::Second))
.and_then(|ts| ts.value().align_to_ceil_by_bucket(time_window_size))
latest_timestamp.map(|ts| time_window_start(ts, time_window_size))
}
/// Finds the active window from the file with the highest sequence number.
@@ -1039,9 +1037,7 @@ fn find_latest_window_in_seconds<'a>(
/// survives the transient state where level 0 is empty right after its files
/// were compacted away. Returns `None` when no file carries a sequence.
///
/// The window key follows the same convention as [`assign_to_windows`]
/// (truncate to seconds, then align up), since it is compared against the
/// window keys produced there.
/// The window key is its inclusive start, as in [`assign_to_windows`].
fn find_active_window_by_sequence<'a>(
files: impl Iterator<Item = &'a FileHandle>,
time_window_size: i64,
@@ -1049,12 +1045,7 @@ fn find_active_window_by_sequence<'a>(
files
.filter(|f| f.meta_ref().sequence.is_some())
.max_by_key(|f| f.meta_ref().sequence)
.and_then(|f| {
f.time_range()
.1
.convert_to(TimeUnit::Second)
.and_then(|ts| ts.value().align_to_ceil_by_bucket(time_window_size))
})
.map(|f| time_window_start(f.time_range().1, time_window_size))
}
#[cfg(test)]
@@ -1347,7 +1338,7 @@ mod tests {
new_file_handle_with_sequence(FileId::random(), 0, 999, 1, 0),
new_file_handle_with_sequence(FileId::random(), 2000, 2999, 0, 0),
];
assert_eq!(Some(3), active_window_of(&files));
assert_eq!(Some(2), active_window_of(&files));
// No sequence and no L0: None, exactly as before.
let files = [new_file_handle_with_sequence(
@@ -1363,7 +1354,7 @@ mod tests {
#[test]
fn test_get_latest_window_in_seconds() {
assert_eq!(
Some(1),
Some(0),
find_latest_window_in_seconds([new_file_handle(FileId::random(), 0, 999, 0)].iter(), 1)
);
assert_eq!(
@@ -1375,7 +1366,7 @@ mod tests {
);
assert_eq!(
Some(-9223372036854000),
Some(-9223372036857600),
find_latest_window_in_seconds(
[new_file_handle(FileId::random(), i64::MIN, i64::MIN + 1, 0)].iter(),
3600,
@@ -1383,7 +1374,7 @@ mod tests {
);
assert_eq!(
(i64::MAX / 10000000 + 1) * 10000,
(i64::MAX / 10000000) * 10000,
find_latest_window_in_seconds(
[new_file_handle(FileId::random(), i64::MIN, i64::MAX, 0)].iter(),
10000,
@@ -1392,7 +1383,7 @@ mod tests {
);
assert_eq!(
Some((i64::MAX / 3600000 + 1) * 3600),
Some((i64::MAX / 3600000) * 3600),
find_latest_window_in_seconds(
[
new_file_handle(FileId::random(), i64::MIN, i64::MAX, 0),
@@ -1404,6 +1395,114 @@ mod tests {
);
}
#[test]
fn test_window_assignment_and_activity_at_boundaries() {
// Before, at, and after positive/negative boundaries, including the first
// fractional second: all picker paths must agree with half-open partitions.
for (unit, ticks_per_second) in [
(TimeUnit::Second, 1),
(TimeUnit::Millisecond, 1_000),
(TimeUnit::Microsecond, 1_000_000),
(TimeUnit::Nanosecond, 1_000_000_000),
] {
for window_size in [1, 3, 86_400] {
for boundary in [-window_size, 0, window_size] {
for (offset, expected_window) in [
(-1, boundary - window_size),
(0, boundary),
(ticks_per_second - 1, boundary),
(ticks_per_second, boundary + i64::from(window_size == 1)),
] {
let timestamp = Timestamp::new(boundary * ticks_per_second + offset, unit);
let file = FileHandle::new(
FileMeta {
time_range: (timestamp, timestamp),
sequence: NonZeroU64::new(1),
..Default::default()
},
crate::test_util::new_noop_file_purger(),
);
let files = [file];
let windows = assign_to_windows(files.iter(), window_size);
assert_eq!(
vec![expected_window],
windows.keys().copied().collect::<Vec<_>>(),
"{timestamp:?}, window size {window_size}"
);
assert_eq!(
Some(expected_window),
find_active_window_by_sequence(files.iter(), window_size)
);
assert_eq!(
Some(expected_window),
find_latest_window_in_seconds(files.iter(), window_size)
);
}
}
}
}
}
#[test]
fn test_window_assignment_at_timestamp_limits() {
for (seconds, expected_window) in [(i64::MIN, i64::MIN), (i64::MAX, i64::MAX - 1)] {
let timestamp = Timestamp::new_second(seconds);
let files = [FileHandle::new(
FileMeta {
time_range: (timestamp, timestamp),
sequence: NonZeroU64::new(1),
..Default::default()
},
crate::test_util::new_noop_file_purger(),
)];
let windows = assign_to_windows(files.iter(), 3);
assert_eq!(
vec![expected_window],
windows.keys().copied().collect::<Vec<_>>()
);
assert_eq!(
Some(expected_window),
find_active_window_by_sequence(files.iter(), 3)
);
assert_eq!(
Some(expected_window),
find_latest_window_in_seconds(files.iter(), 3)
);
assert!(time_window_intersects_range(
expected_window,
3,
&TimestampRange::single(timestamp)
));
}
}
#[tokio::test]
async fn test_midnight_file_is_not_compacted_with_previous_day() {
// Regression for #9441: flush already separates the midnight-only SST.
let midnight = 86_400_000;
CompactionPickerTestCase {
window_size: 86_400,
input_files: [
(midnight - 240_000, midnight - 180_000),
(midnight - 180_000, midnight - 90_000),
(midnight - 90_000, midnight - 30_000),
(midnight, midnight),
]
.into_iter()
.enumerate()
.map(|(index, (start, end))| {
new_file_handle_with_sequence(FileId::random(), start, end, 0, index as u64 + 1)
})
.collect(),
expected_outputs: vec![ExpectedOutput {
input_files: vec![0, 1, 2],
output_level: 1,
}],
}
.check()
.await;
}
#[test]
fn test_assign_to_windows() {
let windows = assign_to_windows(
@@ -1433,7 +1532,7 @@ mod tests {
assert_eq!(
files[0],
windows
.get(&0)
.get(&-3)
.unwrap()
.files()
.next()
@@ -1444,7 +1543,7 @@ mod tests {
assert_eq!(
files[1],
windows
.get(&3)
.get(&0)
.unwrap()
.files()
.next()
@@ -1455,7 +1554,7 @@ mod tests {
assert_eq!(
files[2],
windows
.get(&12)
.get(&9)
.unwrap()
.files()
.next()
@@ -1550,6 +1649,7 @@ mod tests {
.collect::<Vec<_>>();
let windows = assign_to_windows(file_handles.iter(), time_window);
assert_eq!(expected_files.len(), windows.len());
for (expected_window, overlapping, window_files) in expected_files {
let actual_window = windows.get(expected_window).unwrap();
@@ -1574,8 +1674,8 @@ mod tests {
&[(0, 999), (1000, 1999), (2000, 2999)],
2,
&[
(0, false, vec![(0, 999)]),
(2, false, vec![(1000, 1999), (2000, 2999)]),
(0, false, vec![(0, 999), (1000, 1999)]),
(2, false, vec![(2000, 2999)]),
],
);
@@ -1592,9 +1692,8 @@ mod tests {
&[(0, 999), (1000, 1999), (2000, 2999), (3000, 3999)],
2,
&[
(0, false, vec![(0, 999)]),
(2, false, vec![(1000, 1999), (2000, 2999)]),
(4, false, vec![(3000, 3999)]),
(0, false, vec![(0, 999), (1000, 1999)]),
(2, false, vec![(2000, 2999), (3000, 3999)]),
],
);
@@ -1608,9 +1707,8 @@ mod tests {
],
2,
&[
(0, true, vec![(0, 999)]),
(2, true, vec![(1000, 1999), (2000, 2999)]),
(4, true, vec![(0, 3999), (3000, 3999)]),
(0, true, vec![(0, 999), (1000, 1999)]),
(2, true, vec![(0, 3999), (2000, 2999), (3000, 3999)]),
],
);
@@ -1624,43 +1722,39 @@ mod tests {
],
2,
&[
(0, false, vec![(0, 999)]),
(2, true, vec![(1000, 1999), (2000, 2999)]),
(4, true, vec![(1999, 3999), (3000, 3999)]),
(0, true, vec![(0, 999), (1000, 1999)]),
(2, true, vec![(1999, 3999), (2000, 2999), (3000, 3999)]),
],
);
check_assign_to_windows_with_overlapping(
&[
(0, 999), // window 0
(1000, 1999), // window 2
(1000, 1999), // window 0
(2000, 2999), // window 2
(3000, 3999), // window 4
(2999, 3999), // window 4
(3000, 3999), // window 2
(2999, 3999), // window 2
],
2,
&[
// window 2 overlaps with window 4
(0, false, vec![(0, 999)]),
(2, true, vec![(1000, 1999), (2000, 2999)]),
(4, true, vec![(2999, 3999), (3000, 3999)]),
// Overlap within a window does not imply cross-window overlap.
(0, false, vec![(0, 999), (1000, 1999)]),
(2, false, vec![(2000, 2999), (2999, 3999), (3000, 3999)]),
],
);
check_assign_to_windows_with_overlapping(
&[
(0, 999), // window 0
(1000, 1999), // window 2
(1000, 1999), // window 0
(2000, 2999), // window 2
(3000, 3999), // window 4
(0, 1000), // // window 2
(3000, 3999), // window 2
(0, 2000), // window 2
],
2,
&[
// only window 0 overlaps with window 2.
(0, true, vec![(0, 999)]),
(2, true, vec![(0, 1000), (1000, 1999), (2000, 2999)]),
(4, false, vec![(3000, 3999)]),
(0, true, vec![(0, 999), (1000, 1999)]),
(2, true, vec![(0, 2000), (2000, 2999), (3000, 3999)]),
],
);
}
@@ -1680,7 +1774,7 @@ mod tests {
new_file_handle_with_size_sequence_and_primary_key_range(
FileId::random(),
500,
1999,
2999,
0,
2,
10,
@@ -1690,7 +1784,8 @@ mod tests {
let windows = assign_to_windows(files.iter(), 2);
let overlapping = window_has_overlap(windows.get(&2).unwrap(), &windows);
assert_eq!(2, windows.len());
let overlapping = window_has_overlap(windows.get(&0).unwrap(), &windows);
assert!(!overlapping);
}
@@ -2410,7 +2505,7 @@ mod tests {
};
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, Some(9), None)
.build_output_with_time_range(RegionId::from_u64(123), windows, Some(6), None)
.await
.unwrap();
@@ -2427,27 +2522,60 @@ mod tests {
#[test]
fn test_filter_time_windows_by_time_range() {
let time_range = TimestampRange::new(
Timestamp::new_millisecond(1_200),
Timestamp::new_millisecond(1_800),
)
.unwrap();
// Interior, exclusive boundary ends, fractional boundary ends, negative
// timestamps, and empty ranges must all use the same half-open windows.
for (start, end, expected_windows) in [
(1_200, 1_800, vec![0]),
(0, 3_000, vec![0]),
(3_000, 6_000, vec![3]),
(2_999, 3_000, vec![0]),
(2_999, 3_001, vec![0, 3]),
(3_000, 3_001, vec![3]),
(3_001, 3_002, vec![3]),
(-3_000, 0, vec![-3]),
(-3_001, -2_999, vec![-6, -3]),
(0, 0, vec![]),
(1_500, 1_500, vec![]),
] {
let time_range = TimestampRange::with_unit(start, end, TimeUnit::Millisecond).unwrap();
let selected = [-6, -3, 0, 3, 6]
.into_iter()
.filter(|window| time_window_intersects_range(*window, 3, &time_range))
.collect::<Vec<_>>();
assert_eq!(expected_windows, selected, "{time_range}");
}
}
assert!(time_window_intersects_range(3, 3, &time_range));
assert!(!time_window_intersects_range(9, 3, &time_range));
#[test]
fn test_filter_time_windows_with_unbounded_and_extreme_ranges() {
for (range, expected_windows) in [
(TimestampRange::min_to_max(), vec![-3, 0, 3]),
(
TimestampRange::from_start(Timestamp::new_second(0)),
vec![0, 3],
),
(
TimestampRange::until_end(Timestamp::new_second(0), false),
vec![-3],
),
] {
let selected = [-3, 0, 3]
.into_iter()
.filter(|window| time_window_intersects_range(*window, 3, &range))
.collect::<Vec<_>>();
assert_eq!(expected_windows, selected, "{range}");
}
let boundary_range =
TimestampRange::new(Timestamp::new_second(0), Timestamp::new_second(3)).unwrap();
assert!(time_window_intersects_range(0, 3, &boundary_range));
assert!(time_window_intersects_range(3, 3, &boundary_range));
assert!(!time_window_intersects_range(6, 3, &boundary_range));
let overflowing_range = TimestampRange::new(
let upper_range = TimestampRange::new(
Timestamp::new_second(i64::MAX - 1),
Timestamp::new_second(i64::MAX),
)
.unwrap();
assert!(!time_window_intersects_range(0, 4, &overflowing_range));
assert!(!time_window_intersects_range(0, 4, &upper_range));
assert!(time_window_intersects_range(i64::MAX - 3, 4, &upper_range));
let lower_range = TimestampRange::until_end(Timestamp::new_second(i64::MIN), false);
assert!(!time_window_intersects_range(i64::MIN, 3, &lower_range));
}
#[tokio::test]
@@ -2477,7 +2605,7 @@ mod tests {
};
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, Some(9), Some(3))
.build_output_with_time_range(RegionId::from_u64(123), windows, Some(6), Some(3))
.await
.unwrap();
src/mito2/src/read/range_cache.rs
+69 -12
View File
@@ -736,26 +736,33 @@ struct CacheBatchBuffer {
buffered_rows: usize,
buffered_size: usize,
sender: Option<mpsc::UnboundedSender<CacheConcatCommand>>,
/// Handle of the background concat task. Cleared once the task owns a finish
/// command, so only unfinished scans cancel it.
concat_task: Option<common_runtime::JoinHandle<()>>,
}
impl CacheBatchBuffer {
fn new(cache_strategy: &CacheStrategy) -> Self {
let sender = cache_strategy.range_result_memory_limiter().map(|limiter| {
let skip_threshold_bytes = cache_strategy.range_result_cache_size().unwrap_or(0);
let (tx, rx) = mpsc::unbounded_channel();
common_runtime::spawn_query(run_cache_concat_task(
rx,
limiter.clone(),
skip_threshold_bytes,
));
tx
});
let (sender, concat_task) = cache_strategy
.range_result_memory_limiter()
.map(|limiter| {
let skip_threshold_bytes = cache_strategy.range_result_cache_size().unwrap_or(0);
let (tx, rx) = mpsc::unbounded_channel();
let task = common_runtime::spawn_query(run_cache_concat_task(
rx,
limiter.clone(),
skip_threshold_bytes,
));
(tx, task)
})
.unzip();
Self {
buffered_batches: Vec::new(),
buffered_rows: 0,
buffered_size: 0,
sender,
concat_task,
}
}
@@ -814,9 +821,22 @@ impl CacheBatchBuffer {
part_metrics,
result_tx,
})
.is_err()
.is_ok()
{
self.sender = None;
// The task now owns the finish command, so it may keep populating the
// cache after the scan stream is dropped.
self.concat_task = None;
}
}
}
impl Drop for CacheBatchBuffer {
fn drop(&mut self) {
// Still holding the handle means no finish command was sent: the scan was
// cancelled or failed and the queued batches can never reach the cache.
// Aborting releases them even while the task waits for a memory permit.
if let Some(task) = &self.concat_task {
task.abort();
}
}
}
@@ -1570,6 +1590,43 @@ mod tests {
assert_eq!(replayed[1].num_rows(), 1);
}
#[tokio::test]
async fn dropped_cache_buffer_releases_batches_while_waiting_for_memory() {
let strategy = test_cache_strategy();
let limiter = strategy.range_result_memory_limiter().unwrap();
// Take every permit so the concat task blocks before it can compact.
let _permit = limiter
.acquire(limiter.available_permits() * limiter.permit_bytes())
.await
.unwrap();
let batch = make_batch(&vec![1; DEFAULT_READ_BATCH_SIZE / 2 + 1]);
let weak = Arc::downgrade(batch.column(0));
let mut buffer = CacheBatchBuffer::new(&strategy);
buffer.push(batch.clone()).unwrap();
buffer.push(batch).unwrap();
// Both batches were handed to the task instead of staying in the buffer.
assert!(buffer.buffered_batches.is_empty());
// Cancel only once the task is parked on the permit, otherwise the abort
// could land on a task that was never polled.
tokio::time::timeout(std::time::Duration::from_secs(5), async {
while limiter.waited_acquires() == 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("concat task must wait for a memory permit");
drop(buffer);
tokio::time::timeout(std::time::Duration::from_secs(5), async {
while weak.upgrade().is_some() {
tokio::task::yield_now().await;
}
})
.await
.expect("cancelled cache work must release input without waiting for a permit");
}
#[tokio::test]
async fn cache_batch_buffer_finishes_pending_batches() {
let strategy = test_cache_strategy();
src/mito2/src/read/scan_region.rs
+64 -1
View File
@@ -1674,7 +1674,15 @@ impl ScanInput {
};
match maybe_batch {
Some(Ok(batch)) => {
let _ = sender.send(Ok(batch)).await;
// The receiver is gone when the query is cancelled or finishes early,
// so stop reading the source.
if let Err(e) = sender.send(Ok(batch)).await {
debug!(
"Stop parallel scan task, receiver dropped, region_id: {}, error: {}",
region_id, e
);
break;
}
}
Some(Err(e)) => {
let _ = sender.send(Err(e)).await;
@@ -3229,4 +3237,59 @@ mod tests {
.build();
assert!(exact_sequence_range(&request, &version).is_err());
}
#[tokio::test]
async fn test_parallel_flat_source_stops_after_receiver_dropped() {
use std::sync::atomic::{AtomicUsize, Ordering};
use datatypes::arrow::array::Int64Array;
const SOURCE_BATCHES: usize = 10_000;
let metadata = Arc::new(metadata_with_primary_key(vec![0, 1], false));
let mapper = FlatProjectionMapper::new(&metadata, [0, 2, 3]).unwrap();
let env = SchedulerEnv::new().await;
let input = ScanInput::builder(env.access_layer.clone(), mapper).build();
let schema = Arc::new(ArrowSchema::new(vec![Field::new(
"v",
ArrowDataType::Int64,
false,
)]));
let batch =
RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(vec![1]))]).unwrap();
let pulled = Arc::new(AtomicUsize::new(0));
let mut released = Vec::new();
let sources = (0..2)
.map(|_| {
let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
released.push(release_rx);
let batch = batch.clone();
let pulled = pulled.clone();
futures::stream::iter(0..SOURCE_BATCHES)
.map(move |_| {
// Dropped together with the source stream.
let _ = &release_tx;
pulled.fetch_add(1, Ordering::Relaxed);
Ok(batch.clone())
})
.boxed()
})
.collect();
let streams = input
.create_parallel_flat_sources(sources, Arc::new(Semaphore::new(2)), 1)
.unwrap();
drop(streams);
for release_rx in released {
tokio::time::timeout(std::time::Duration::from_secs(10), release_rx)
.await
.unwrap()
.unwrap_err();
}
// Each task may have read a few batches ahead into its channel before the
// receiver went away, but must not drain the rest of its source.
assert!(pulled.load(Ordering::Relaxed) < 100);
}
}